feat(diagnostics): add SQL execution debug logging
This commit is contained in:
Abeautifulsnow
GitHub
parent
fc1f02a931
commit
badb3b61b9
|
|
@ -49,6 +49,7 @@ pub mod schema;
|
|||
pub mod schema_diff;
|
||||
pub mod sql;
|
||||
pub mod sql_analysis;
|
||||
pub mod sql_diagnostics;
|
||||
pub mod sql_dialect;
|
||||
pub mod sql_editability;
|
||||
pub mod sql_file_import;
|
||||
|
|
|
|||
|
|
@ -34,6 +34,8 @@ use crate::sql::{split_sql_batches, split_sql_statements, starts_with_executable
|
|||
pub const QUERY_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
pub const MAX_ROWS: usize = 10000;
|
||||
pub const QUERY_CANCELED: &str = "Query canceled";
|
||||
const SQL_OMITTED_ERROR_CONTEXT: &str =
|
||||
"SQL text omitted from user-facing error; enable debug SQL diagnostics for a redacted statement.";
|
||||
#[cfg(feature = "duckdb-bundled")]
|
||||
const DUCKDB_INTERRUPT_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
|
||||
#[cfg(feature = "duckdb-bundled")]
|
||||
|
|
@ -46,6 +48,14 @@ pub enum PoolErrorAction {
|
|||
ReconnectAndRetry,
|
||||
}
|
||||
|
||||
fn query_error_with_omitted_sql_context(error: &str, _sql: &str) -> String {
|
||||
if error.contains(SQL_OMITTED_ERROR_CONTEXT) {
|
||||
error.to_string()
|
||||
} else {
|
||||
format!("{error}\n{SQL_OMITTED_ERROR_CONTEXT}")
|
||||
}
|
||||
}
|
||||
|
||||
/// A multi-statement result with metadata intended for query clients.
|
||||
///
|
||||
/// `execution_error` is emitted for synthesized per-statement errors so clients
|
||||
|
|
@ -1194,6 +1204,7 @@ pub async fn do_execute(
|
|||
cancel_token: Option<CancellationToken>,
|
||||
options: QueryExecutionOptions,
|
||||
) -> Result<db::QueryResult, String> {
|
||||
crate::sql_diagnostics::debug_sql("do_execute", sql);
|
||||
if let Some(execution_id) = options.execution_id.as_deref() {
|
||||
state.running_queries.set_pool_key(execution_id, pool_key.to_string());
|
||||
}
|
||||
|
|
@ -1722,11 +1733,15 @@ pub async fn execute_sql_statement_with_options(
|
|||
// on that tab-scoped pool so connection-level state (for example MySQL @vars)
|
||||
// survives across runs.
|
||||
let pool_key = if database.is_empty() {
|
||||
state.get_or_create_pool_for_session(connection_id, None, options.client_session_id.as_deref()).await?
|
||||
state
|
||||
.get_or_create_pool_for_session(connection_id, None, options.client_session_id.as_deref())
|
||||
.await
|
||||
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?
|
||||
} else {
|
||||
state
|
||||
.get_or_create_pool_for_session(connection_id, Some(database), options.client_session_id.as_deref())
|
||||
.await?
|
||||
.await
|
||||
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?
|
||||
};
|
||||
|
||||
if is_canceled(&cancel_token) {
|
||||
|
|
@ -1738,19 +1753,26 @@ pub async fn execute_sql_statement_with_options(
|
|||
do_execute(state, &pool_key, mysql_dialect, Some(database), sql, schema, cancel_token.clone(), options.clone())
|
||||
.await;
|
||||
|
||||
let with_sql_context =
|
||||
|r: Result<db::QueryResult, String>| r.map_err(|e| query_error_with_omitted_sql_context(&e, sql));
|
||||
|
||||
let action = result.as_ref().err().map(|e| query_pool_error_action(db_type, sql, e));
|
||||
match action {
|
||||
Some(PoolErrorAction::ReconnectAndRetry) if !is_canceled(&cancel_token) => {
|
||||
let db_opt = if database.is_empty() { None } else { Some(database) };
|
||||
let new_key =
|
||||
state.reconnect_pool_for_session(connection_id, db_opt, options.client_session_id.as_deref()).await?;
|
||||
do_execute(state, &new_key, mysql_dialect, Some(database), sql, schema, cancel_token, options).await
|
||||
let new_key = state
|
||||
.reconnect_pool_for_session(connection_id, db_opt, options.client_session_id.as_deref())
|
||||
.await
|
||||
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?;
|
||||
with_sql_context(
|
||||
do_execute(state, &new_key, mysql_dialect, Some(database), sql, schema, cancel_token, options).await,
|
||||
)
|
||||
}
|
||||
Some(PoolErrorAction::Discard) => {
|
||||
state.remove_pool_by_key(&pool_key).await;
|
||||
result
|
||||
with_sql_context(result)
|
||||
}
|
||||
_ => result,
|
||||
_ => with_sql_context(result),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1767,7 +1789,8 @@ async fn execute_postgres_drop_database(
|
|||
let admin_database = postgres_drop_database_admin_database(target_database);
|
||||
let pool_key = state
|
||||
.get_or_create_pool_for_session(connection_id, Some(admin_database), options.client_session_id.as_deref())
|
||||
.await?;
|
||||
.await
|
||||
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?;
|
||||
if let Some(execution_id) = options.execution_id.as_deref() {
|
||||
state.running_queries.set_pool_key(execution_id, pool_key.clone());
|
||||
}
|
||||
|
|
@ -1915,11 +1938,15 @@ pub async fn execute_multi_core_with_options_for_client(
|
|||
}
|
||||
|
||||
let pool_key = if database.is_empty() {
|
||||
state.get_or_create_pool_for_session(connection_id, None, options.client_session_id.as_deref()).await?
|
||||
state
|
||||
.get_or_create_pool_for_session(connection_id, None, options.client_session_id.as_deref())
|
||||
.await
|
||||
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?
|
||||
} else {
|
||||
state
|
||||
.get_or_create_pool_for_session(connection_id, Some(database), options.client_session_id.as_deref())
|
||||
.await?
|
||||
.await
|
||||
.map_err(|e| query_error_with_omitted_sql_context(&e, sql))?
|
||||
};
|
||||
if let Some(execution_id) = options.execution_id.as_deref() {
|
||||
state.running_queries.set_pool_key(execution_id, pool_key.clone());
|
||||
|
|
@ -2304,10 +2331,14 @@ pub async fn execute_statements(
|
|||
schema: Option<&str>,
|
||||
timeout_secs: Option<u64>,
|
||||
) -> Result<db::QueryResult, String> {
|
||||
let sql_ctx = statements.first().map(|s| s.as_str()).unwrap_or("");
|
||||
let pool_key = if database.is_empty() {
|
||||
connection_id.to_string()
|
||||
} else {
|
||||
state.get_or_create_pool(connection_id, Some(database)).await?
|
||||
state
|
||||
.get_or_create_pool(connection_id, Some(database))
|
||||
.await
|
||||
.map_err(|e| query_error_with_omitted_sql_context(&e, sql_ctx))?
|
||||
};
|
||||
|
||||
let mut total_affected: u64 = 0;
|
||||
|
|
@ -2352,7 +2383,7 @@ pub async fn execute_statements(
|
|||
}
|
||||
PoolErrorAction::Keep => {}
|
||||
}
|
||||
return Err(err);
|
||||
return Err(query_error_with_omitted_sql_context(&err, sql_ctx));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -2385,11 +2416,9 @@ pub async fn execute_statements(
|
|||
}
|
||||
PoolErrorAction::Keep => {}
|
||||
}
|
||||
return Err(format!(
|
||||
"Statement {} failed: {}. Previous {} statement(s) may have been committed.",
|
||||
i + 1,
|
||||
e,
|
||||
i
|
||||
return Err(query_error_with_omitted_sql_context(
|
||||
&format!("Statement {} failed: {}. Previous {} statement(s) may have been committed.", i + 1, e, i),
|
||||
sql,
|
||||
));
|
||||
}
|
||||
}
|
||||
|
|
@ -2427,10 +2456,14 @@ pub async fn execute_statements_in_transaction(
|
|||
statements: &[String],
|
||||
schema: Option<&str>,
|
||||
) -> Result<db::QueryResult, String> {
|
||||
let sql_ctx = statements.first().map(|s| s.as_str()).unwrap_or("");
|
||||
let pool_key = if database.is_empty() {
|
||||
connection_id.to_string()
|
||||
} else {
|
||||
state.get_or_create_pool(connection_id, Some(database)).await?
|
||||
state
|
||||
.get_or_create_pool(connection_id, Some(database))
|
||||
.await
|
||||
.map_err(|e| query_error_with_omitted_sql_context(&e, sql_ctx))?
|
||||
};
|
||||
|
||||
// Read-only check: intercept all transaction paths before dispatching
|
||||
|
|
@ -2607,7 +2640,7 @@ async fn exec_tx_pg_statements(
|
|||
async { tx.execute(sql, &[]).await.map_err(|e| e.to_string()) },
|
||||
)
|
||||
.await
|
||||
.map_err(|e| format!("Statement {} failed: {}", i + 1, e))?;
|
||||
.map_err(|e| query_error_with_omitted_sql_context(&format!("Statement {} failed: {}", i + 1, e), sql))?;
|
||||
total_affected += affected;
|
||||
}
|
||||
tokio::time::timeout(budget.cleanup_timeout, tx.commit())
|
||||
|
|
@ -2641,7 +2674,7 @@ async fn exec_tx_mysql_inner(
|
|||
Err(e) => {
|
||||
let _ = mysql_query_drop_with_timeout(&mut conn, "ROLLBACK", budget.cleanup_timeout, "ROLLBACK failed")
|
||||
.await;
|
||||
return Err(format!("Statement {} failed: {}", i + 1, e));
|
||||
return Err(query_error_with_omitted_sql_context(&format!("Statement {} failed: {}", i + 1, e), sql));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -2701,7 +2734,10 @@ async fn exec_tx_sqlite_inner(
|
|||
Ok(_) => total_affected += conn.changes(),
|
||||
Err(e) => {
|
||||
let _ = conn.execute_batch("ROLLBACK");
|
||||
return Err(format!("Statement {} failed: {}", i + 1, e));
|
||||
return Err(query_error_with_omitted_sql_context(
|
||||
&format!("Statement {} failed: {}", i + 1, e),
|
||||
sql,
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -2786,7 +2822,7 @@ async fn exec_tx_explicit_inner(
|
|||
{
|
||||
log::error!("ROLLBACK failed after statement {} error: {}", i + 1, rb_err);
|
||||
}
|
||||
return Err(format!("Statement {} failed: {}", i + 1, e));
|
||||
return Err(query_error_with_omitted_sql_context(&format!("Statement {} failed: {}", i + 1, e), sql));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -2819,7 +2855,7 @@ async fn exec_tx_none_inner(
|
|||
) -> Result<db::QueryResult, String> {
|
||||
let mut total_affected: u64 = 0;
|
||||
for (i, sql) in statements.iter().enumerate() {
|
||||
log::info!("[query][tx-none:statement:start] index={} sql={}", i + 1, sql);
|
||||
log::info!("[query][tx-none:statement:start] index={}", i + 1);
|
||||
match do_execute(state, pool_key, mysql_dialect, database, sql, schema, None, QueryExecutionOptions::default())
|
||||
.await
|
||||
{
|
||||
|
|
@ -2829,10 +2865,9 @@ async fn exec_tx_none_inner(
|
|||
}
|
||||
Err(e) => {
|
||||
log::warn!("Statement {} failed (no transaction support): {}", i + 1, e);
|
||||
return Err(format!(
|
||||
"Statement {} failed: {}. No transaction support for this database type.",
|
||||
i + 1,
|
||||
e
|
||||
return Err(query_error_with_omitted_sql_context(
|
||||
&format!("Statement {} failed: {}. No transaction support for this database type.", i + 1, e),
|
||||
sql,
|
||||
));
|
||||
}
|
||||
}
|
||||
|
|
@ -4043,6 +4078,62 @@ mod tests {
|
|||
assert!(is_connection_error("Error occurred while creating a new object: error communicating with the server"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn query_error_context_omits_raw_sql_and_is_not_duplicated() {
|
||||
let sql = "select 'secret-123' as token";
|
||||
let error = query_error_with_omitted_sql_context("driver rejected statement", sql);
|
||||
|
||||
assert!(error.contains("driver rejected statement"));
|
||||
assert!(error.contains(SQL_OMITTED_ERROR_CONTEXT));
|
||||
assert!(!error.contains("secret-123"));
|
||||
assert!(!error.contains("SQL:"));
|
||||
|
||||
let repeated = query_error_with_omitted_sql_context(&error, sql);
|
||||
assert_eq!(repeated.matches(SQL_OMITTED_ERROR_CONTEXT).count(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reconnect_retry_error_context_omits_raw_sql() {
|
||||
let sql = "select 'secret-123' as token";
|
||||
let reconnect_error = query_error_with_omitted_sql_context("connection reset after reconnect", sql);
|
||||
|
||||
assert!(reconnect_error.contains("connection reset after reconnect"));
|
||||
assert!(reconnect_error.contains(SQL_OMITTED_ERROR_CONTEXT));
|
||||
assert!(!reconnect_error.contains("secret-123"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn execute_statements_error_omits_raw_sql() {
|
||||
let sql = "select 'secret-token' as t";
|
||||
let err = query_error_with_omitted_sql_context(
|
||||
&format!(
|
||||
"Statement {} failed: {}. Previous {} statement(s) may have been committed.",
|
||||
2, "driver error", 1
|
||||
),
|
||||
sql,
|
||||
);
|
||||
|
||||
assert!(err.contains("driver error"));
|
||||
assert!(err.contains(SQL_OMITTED_ERROR_CONTEXT));
|
||||
assert!(!err.contains("secret-token"));
|
||||
assert!(!err.contains("SQL:"));
|
||||
assert!(err.contains("Statement 2 failed:"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_transaction_error_omits_raw_sql() {
|
||||
let sql = "delete from users where id = 'secret-id'";
|
||||
let err = query_error_with_omitted_sql_context(
|
||||
&format!("Statement {} failed: {}. No transaction support for this database type.", 3, "batch error"),
|
||||
sql,
|
||||
);
|
||||
|
||||
assert!(err.contains("batch error"));
|
||||
assert!(err.contains(SQL_OMITTED_ERROR_CONTEXT));
|
||||
assert!(!err.contains("secret-id"));
|
||||
assert!(err.contains("Statement 3 failed:"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn is_connection_error_detects_oracle_idle_timeout() {
|
||||
assert!(is_connection_error("ORA-02396: exceeded maximum idle time, please connect again"));
|
||||
|
|
|
|||
|
|
@ -0,0 +1,263 @@
|
|||
const DEFAULT_SQL_DIAGNOSTIC_MAX_CHARS: usize = 512;
|
||||
|
||||
fn is_sensitive_key(key: &str) -> bool {
|
||||
let key = key.to_ascii_lowercase();
|
||||
key.contains("password")
|
||||
|| key.contains("passwd")
|
||||
|| key == "pwd"
|
||||
|| key.contains("secret")
|
||||
|| key.contains("token")
|
||||
|| key.contains("api_key")
|
||||
|| key.contains("apikey")
|
||||
|| key.contains("access_key")
|
||||
|| key.contains("private_key")
|
||||
|| key.contains("credential")
|
||||
|| key.contains("authorization")
|
||||
|| key.contains("bearer")
|
||||
}
|
||||
|
||||
fn truncate_for_diagnostics(value: String, max_chars: usize, input_truncated: bool) -> String {
|
||||
if value.chars().count() <= max_chars {
|
||||
return if input_truncated { format!("{value}…[truncated]") } else { value };
|
||||
}
|
||||
let head: String = value.chars().take(max_chars).collect();
|
||||
format!("{head}…[truncated]")
|
||||
}
|
||||
|
||||
fn bounded_input(sql: &str, max_chars: usize) -> (&str, bool) {
|
||||
if max_chars == 0 {
|
||||
return ("", !sql.is_empty());
|
||||
}
|
||||
match sql.char_indices().nth(max_chars) {
|
||||
Some((index, _)) => (&sql[..index], true),
|
||||
None => (sql, false),
|
||||
}
|
||||
}
|
||||
|
||||
fn redact_literals(sql: &str) -> String {
|
||||
let chars: Vec<char> = sql.chars().collect();
|
||||
let mut out = String::new();
|
||||
let mut i = 0;
|
||||
while i < chars.len() {
|
||||
let ch = chars[i];
|
||||
let next = chars.get(i + 1).copied();
|
||||
if matches!(ch, '\'' | '"' | '`') {
|
||||
out.push(ch);
|
||||
out.push_str("[REDACTED]");
|
||||
out.push(ch);
|
||||
i += 1;
|
||||
while i < chars.len() {
|
||||
let current = chars[i];
|
||||
if current == ch {
|
||||
if chars.get(i + 1).copied() == Some(ch) {
|
||||
i += 2;
|
||||
continue;
|
||||
}
|
||||
i += 1;
|
||||
break;
|
||||
}
|
||||
if current == '\\' && ch != '`' {
|
||||
i += 2;
|
||||
} else {
|
||||
i += 1;
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if ch == '$' {
|
||||
let j = i + 1;
|
||||
if j >= chars.len() {
|
||||
out.push(ch);
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
if chars[j] == '$' {
|
||||
// $$...$$ dollar-quoted string
|
||||
out.push_str("$$[REDACTED]$$");
|
||||
i += 2;
|
||||
while i + 1 < chars.len() && !(chars[i] == '$' && chars[i + 1] == '$') {
|
||||
i += 1;
|
||||
}
|
||||
if i + 1 < chars.len() {
|
||||
i += 2;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
// $tag$...$tag$ dollar-quoted string
|
||||
let tag_start = j;
|
||||
let mut tag_end = j;
|
||||
while tag_end < chars.len() && (chars[tag_end].is_ascii_alphanumeric() || chars[tag_end] == '_') {
|
||||
tag_end += 1;
|
||||
}
|
||||
if tag_end > tag_start && tag_end < chars.len() && chars[tag_end] == '$' {
|
||||
let tag: String = chars[tag_start..tag_end].iter().collect();
|
||||
out.push_str("$[REDACTED]$");
|
||||
i = tag_end + 1;
|
||||
let closing: Vec<char> = format!("${}$", tag).chars().collect();
|
||||
while i + closing.len() <= chars.len() {
|
||||
if chars[i..i + closing.len()] == closing[..] {
|
||||
i += closing.len();
|
||||
break;
|
||||
}
|
||||
i += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
out.push(ch);
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
if ch == '-' && next == Some('-') {
|
||||
out.push_str("--[REDACTED_COMMENT]");
|
||||
i += 2;
|
||||
while i < chars.len() && chars[i] != '\n' && chars[i] != '\r' {
|
||||
i += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if ch == '/' && next == Some('*') {
|
||||
out.push_str("/*[REDACTED_COMMENT]*/");
|
||||
i += 2;
|
||||
while i + 1 < chars.len() && !(chars[i] == '*' && chars[i + 1] == '/') {
|
||||
i += 1;
|
||||
}
|
||||
if i + 1 < chars.len() {
|
||||
i += 2;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
out.push(ch);
|
||||
i += 1;
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn redact_sensitive_assignments(sql: &str) -> String {
|
||||
let chars: Vec<char> = sql.chars().collect();
|
||||
let mut out = String::new();
|
||||
let mut i = 0;
|
||||
while i < chars.len() {
|
||||
if chars[i].is_whitespace() {
|
||||
out.push(chars[i]);
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
let start = i;
|
||||
while i < chars.len()
|
||||
&& (chars[i].is_ascii_alphanumeric() || chars[i] == '_' || chars[i] == '-' || chars[i] == '.')
|
||||
{
|
||||
i += 1;
|
||||
}
|
||||
if i == start {
|
||||
out.push(chars[i]);
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
let key: String = chars[start..i].iter().collect();
|
||||
let mut j = i;
|
||||
while j < chars.len() && chars[j].is_whitespace() {
|
||||
j += 1;
|
||||
}
|
||||
if j < chars.len() && (chars[j] == '=' || chars[j] == ':') {
|
||||
if is_sensitive_key(&key) {
|
||||
out.push_str(&key);
|
||||
for k in i..j {
|
||||
out.push(chars[k]);
|
||||
}
|
||||
out.push(chars[j]);
|
||||
j += 1;
|
||||
while j < chars.len() && chars[j].is_whitespace() {
|
||||
out.push(chars[j]);
|
||||
j += 1;
|
||||
}
|
||||
while j < chars.len() && !chars[j].is_whitespace() {
|
||||
j += 1;
|
||||
}
|
||||
out.push_str("[REDACTED]");
|
||||
i = j;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
out.push_str(&key);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
pub fn redact_sql_for_diagnostics(sql: &str) -> String {
|
||||
let max_chars = DEFAULT_SQL_DIAGNOSTIC_MAX_CHARS;
|
||||
let (bounded_sql, input_truncated) = bounded_input(sql, max_chars);
|
||||
truncate_for_diagnostics(redact_sensitive_assignments(&redact_literals(bounded_sql)), max_chars, input_truncated)
|
||||
}
|
||||
|
||||
pub fn debug_sql(scope: &str, sql: &str) {
|
||||
log::debug!("[{scope}] sql={}", redact_sql_for_diagnostics(sql));
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn redacts_sensitive_literals_and_bounds_large_sql() {
|
||||
let sql = format!(
|
||||
"select * from users where password = 'secret-123' and api_key=abc and name = 'alice' {};",
|
||||
"x".repeat(900)
|
||||
);
|
||||
let redacted = redact_sql_for_diagnostics(&sql);
|
||||
assert!(!redacted.contains("secret-123"));
|
||||
assert!(!redacted.contains("api_key=abc"));
|
||||
assert!(!redacted.contains("alice"));
|
||||
// Literals not part of sensitive assignments retain single-quote redaction
|
||||
assert!(redacted.contains("'[REDACTED]'"), "name literal should be redacted: {}", redacted);
|
||||
// Sensitive assignment values are redacted with bracket notation
|
||||
assert!(redacted.contains("password = [REDACTED]"));
|
||||
assert!(redacted.contains("api_key=[REDACTED]"));
|
||||
assert!(redacted.contains("truncated"));
|
||||
assert!(redacted.len() < sql.len());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn redacts_space_separated_sensitive_assignments() {
|
||||
let sql = "select * from users where password = hunter2";
|
||||
let redacted = redact_sensitive_assignments(sql);
|
||||
assert!(!redacted.contains("hunter2"));
|
||||
assert!(redacted.contains("password = [REDACTED]"));
|
||||
assert!(redacted.contains("select"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn redacts_dollar_quoted_strings() {
|
||||
let sql = "select $$secret$$, $tag$hello$tag$ from t";
|
||||
let redacted = redact_sql_for_diagnostics(sql);
|
||||
assert!(redacted.contains("$$[REDACTED]$$"));
|
||||
assert!(redacted.contains("$[REDACTED]$"));
|
||||
assert!(!redacted.contains("secret"));
|
||||
assert!(!redacted.contains("hello"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn large_input_bounded_allocation() {
|
||||
let sql = "x".repeat(1_000_000);
|
||||
let redacted = redact_sql_for_diagnostics(&sql);
|
||||
assert!(redacted.len() <= 550);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncation_inside_unclosed_literal_does_not_leak_prefix() {
|
||||
let sql = format!("select '{}'", "secret-".repeat(1_000));
|
||||
let redacted = truncate_for_diagnostics(redact_literals(bounded_input(&sql, 32).0), 32, true);
|
||||
assert!(!redacted.contains("secret-"));
|
||||
assert!(redacted.contains("[REDACTED]"));
|
||||
assert!(redacted.contains("truncated"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn truncation_inside_sensitive_assignment_does_not_leak_prefix() {
|
||||
let sql = format!("password = {}", "secret-token".repeat(1_000));
|
||||
let (bounded, truncated) = bounded_input(&sql, 24);
|
||||
let redacted = truncate_for_diagnostics(redact_sensitive_assignments(&redact_literals(bounded)), 24, truncated);
|
||||
assert!(!redacted.contains("secret-token"));
|
||||
assert!(redacted.contains("password = [REDACTED]"));
|
||||
assert!(redacted.contains("truncated"));
|
||||
}
|
||||
}
|
||||
|
|
@ -23,7 +23,7 @@ serde_json = "1.0"
|
|||
uuid = { version = "1", features = ["v4"] }
|
||||
argon2 = "0.5"
|
||||
log = "0.4"
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter", "tracing-log"] }
|
||||
tracing = "0.1"
|
||||
async-stream = "0.3"
|
||||
futures = "0.3"
|
||||
|
|
|
|||
|
|
@ -320,6 +320,8 @@ pub async fn execute_query(
|
|||
);
|
||||
let cancel_token = registered.token();
|
||||
|
||||
tracing::debug!(connection_id = %req.connection_id, "execute_query");
|
||||
|
||||
let result = dbx_core::query::execute_sql_statement_with_options(
|
||||
&state.app,
|
||||
&req.connection_id,
|
||||
|
|
@ -359,6 +361,8 @@ pub async fn execute_multi(
|
|||
);
|
||||
let cancel_token = registered.token();
|
||||
|
||||
tracing::debug!(connection_id = %req.connection_id, "execute_multi");
|
||||
|
||||
let result = dbx_core::query::execute_multi_core_with_options_for_client(
|
||||
&state.app,
|
||||
&req.connection_id,
|
||||
|
|
@ -390,6 +394,7 @@ pub async fn execute_batch(
|
|||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<ExecuteBatchRequest>,
|
||||
) -> Result<Json<dbx_core::db::QueryResult>, AppError> {
|
||||
tracing::debug!(connection_id = %req.connection_id, "execute_batch");
|
||||
let result = dbx_core::query::execute_statements(
|
||||
&state.app,
|
||||
&req.connection_id,
|
||||
|
|
@ -447,6 +452,7 @@ pub async fn execute_script(
|
|||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<ExecuteQueryRequest>,
|
||||
) -> Result<Json<dbx_core::db::QueryResult>, AppError> {
|
||||
tracing::debug!(connection_id = %req.connection_id, "execute_script");
|
||||
let db_type = {
|
||||
let configs = state.app.configs.read().await;
|
||||
configs.get(&req.connection_id).map(|config| config.db_type)
|
||||
|
|
@ -472,6 +478,7 @@ pub async fn execute_in_transaction(
|
|||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<ExecuteBatchRequest>,
|
||||
) -> Result<Json<dbx_core::db::QueryResult>, AppError> {
|
||||
tracing::debug!(connection_id = %req.connection_id, "execute_in_transaction");
|
||||
let result = dbx_core::query::execute_statements_in_transaction(
|
||||
&state.app,
|
||||
&req.connection_id,
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
#!/usr/bin/env node
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { buildSchemaContext, createBackend, DIRECT_QUERY_TYPES, BRIDGE_REQUIRED_TYPES, evaluateSqlSafety, formatSchemaContext, getDbxDiagnostics, isMainModule, postBridge, type Backend, type DbxDiagnostics, type SqlSafetyOptions } from "@dbx-app/node-core";
|
||||
import { buildSchemaContext, createBackend, DIRECT_QUERY_TYPES, BRIDGE_REQUIRED_TYPES, evaluateSqlSafety, formatSchemaContext, getDbxDiagnostics, isMainModule, postBridge, supportsHashLineComments, type Backend, type DbxDiagnostics, type SqlSafetyOptions } from "@dbx-app/node-core";
|
||||
import { connectionSummary, csvTable, errorPayload, formatCell, formatErrorMessage, mdTable } from "./cli-format.js";
|
||||
|
||||
export interface CliResult {
|
||||
|
|
@ -165,6 +165,7 @@ export async function runCli(argv: string[], options: RunOptions = {}): Promise<
|
|||
}
|
||||
const sqlArg = usesDefaultConnection ? args[1] : args[2];
|
||||
const sql = flags.file ? await readFile(flags.file, "utf-8") : required(sqlArg, "SQL string or --file is required.");
|
||||
const config = await findConnectionOrThrow(backend, connectionName);
|
||||
const envSafety = sqlSafetyFromCliEnv(env);
|
||||
if (flags.allowDangerous && !flags.allowWrites && !envSafety.allowWrites) {
|
||||
throw new CliError("INVALID_OPTION", "--allow-dangerous-sql requires --allow-writes.");
|
||||
|
|
@ -172,10 +173,10 @@ export async function runCli(argv: string[], options: RunOptions = {}): Promise<
|
|||
const safetyOptions: SqlSafetyOptions = {
|
||||
allowWrites: flags.allowWrites || envSafety.allowWrites,
|
||||
allowDangerous: flags.allowDangerous || envSafety.allowDangerous,
|
||||
hashLineComments: supportsHashLineComments(config.db_type),
|
||||
};
|
||||
const safety = evaluateSqlSafety(sql, safetyOptions);
|
||||
if (!safety.allowed) return fail("SQL_BLOCKED", safety.reason ?? "SQL blocked.", flags.json);
|
||||
const config = await findConnectionOrThrow(backend, connectionName);
|
||||
const result = await backend.executeQuery(config, sql, { maxRows: flags.maxRows, timeoutMs: flags.timeoutMs });
|
||||
if (flags.format === "json") {
|
||||
return okJson({ connection: connectionName, columns: result.columns, rows: result.rows, row_count: result.row_count });
|
||||
|
|
|
|||
|
|
@ -124,6 +124,10 @@ DBX_MCP_ALLOW_DANGEROUS_SQL=1
|
|||
|
||||
Redis connections use `dbx_execute_redis_command` instead of `dbx_execute_query`. Redis write commands honor `DBX_MCP_ALLOW_WRITES`; dangerous Redis commands such as `KEYS`, `FLUSHALL`, and `EVAL` require `DBX_MCP_ALLOW_DANGEROUS_SQL=1`.
|
||||
|
||||
## SQL Diagnostics Privacy
|
||||
|
||||
SQL statements are not included in normal MCP errors and are not logged by default. To enable temporary diagnostics, set `DBX_MCP_DEBUG_SQL=1` (or `DBX_SQL_DEBUG=1`). Diagnostic statements redact quoted literals and common secret assignments, and are truncated to 512 characters. Do not enable this setting unless the resulting diagnostic metadata is appropriate for the environment.
|
||||
|
||||
## How It Works
|
||||
|
||||
```
|
||||
|
|
|
|||
|
|
@ -19,8 +19,10 @@ import {
|
|||
isLikelyMongoMutation,
|
||||
isProductionDatabase,
|
||||
postBridge,
|
||||
logSqlDiagnostic,
|
||||
sqlSafetyFromEnv,
|
||||
splitSqlStatements,
|
||||
supportsHashLineComments,
|
||||
type Backend,
|
||||
type ConnectionConfig,
|
||||
type QueryResult,
|
||||
|
|
@ -231,6 +233,7 @@ export function createDbxMcpServer(backend: Backend, options: { isWebMode?: bool
|
|||
sql: z.string().describe("SQL query to execute"),
|
||||
},
|
||||
async ({ connection_id, connection_name, database, sql }) => {
|
||||
logSqlDiagnostic("dbx_execute_query", sql, { connection_id, connection_name, database });
|
||||
const { config, error } = await resolveConnection(backend, scope, connection_id, connection_name);
|
||||
if (error) return error;
|
||||
const scopedConfig = config!;
|
||||
|
|
@ -238,7 +241,8 @@ export function createDbxMcpServer(backend: Backend, options: { isWebMode?: bool
|
|||
return toolError("REDIS_COMMAND_REQUIRED", "Redis connections do not accept SQL through dbx_execute_query. Use dbx_execute_redis_command with a Redis command such as GET key or INFO.");
|
||||
}
|
||||
if (scopedConfig.db_type !== "mongodb") {
|
||||
const safety = evaluateSqlSafety(sql, { ...sqlSafetyFromEnv(), allowMultipleStatements: true });
|
||||
const hashLineComments = supportsHashLineComments(scopedConfig.db_type);
|
||||
const safety = evaluateSqlSafety(sql, { ...sqlSafetyFromEnv(), allowMultipleStatements: true, hashLineComments });
|
||||
if (!safety.allowed) return toolError("SQL_BLOCKED", safety.reason ?? "SQL blocked.");
|
||||
const production = assessProductionSql(sql, scopedConfig, database ?? scope.database ?? scopedConfig.database);
|
||||
if (production.active && production.isMutation) {
|
||||
|
|
@ -250,7 +254,7 @@ export function createDbxMcpServer(backend: Backend, options: { isWebMode?: bool
|
|||
// MongoDB shell commands don't fit the SQL safety evaluator; the backend
|
||||
// (node-core executeQuery) applies command-aware read/write gating.
|
||||
try {
|
||||
const statements = scopedConfig.db_type === "mongodb" ? [sql] : splitSqlStatements(sql);
|
||||
const statements = scopedConfig.db_type === "mongodb" ? [sql] : splitSqlStatements(sql, { hashLineComments: supportsHashLineComments(scopedConfig.db_type) });
|
||||
const results = [];
|
||||
for (const statement of statements) {
|
||||
results.push(await backend.executeQuery(withDatabase(scopedConfig, database ?? scope.database), statement));
|
||||
|
|
@ -479,7 +483,8 @@ export function createDbxMcpServer(backend: Backend, options: { isWebMode?: bool
|
|||
if (!safety.allowed) return toolError("SQL_BLOCKED", safety.reason ?? "Query blocked.");
|
||||
}
|
||||
} else {
|
||||
const safety = evaluateSqlSafety(sql, { ...safetyOptions, allowMultipleStatements: true });
|
||||
const hashLineComments = supportsHashLineComments(config?.db_type);
|
||||
const safety = evaluateSqlSafety(sql, { ...safetyOptions, allowMultipleStatements: true, hashLineComments });
|
||||
if (!safety.allowed) return toolError("SQL_BLOCKED", safety.reason ?? "SQL blocked.");
|
||||
}
|
||||
if (config?.db_type === "mongodb") {
|
||||
|
|
@ -494,6 +499,7 @@ export function createDbxMcpServer(backend: Backend, options: { isWebMode?: bool
|
|||
}
|
||||
// MongoDB shell commands bypass the SQL safety evaluator; pass MCP
|
||||
// safety flags to the desktop executor for command-aware gating.
|
||||
logSqlDiagnostic("dbx_execute_in_app", sql, { connection_id: config!.id, connection_name: config!.name, database });
|
||||
return bridgeRequest(
|
||||
"/execute-query",
|
||||
{
|
||||
|
|
|
|||
|
|
@ -256,6 +256,52 @@ test("redis command tool executes redis commands on the selected database", asyn
|
|||
assert.match(result.content[0].text, /value-1/);
|
||||
});
|
||||
|
||||
test("dbx_execute_query does not log SQL when debug diagnostics are disabled", async () => {
|
||||
const original = console.error;
|
||||
const originalDebug = process.env.DBX_SQL_DEBUG;
|
||||
const originalMcpDebug = process.env.DBX_MCP_DEBUG_SQL;
|
||||
const messages: unknown[][] = [];
|
||||
delete process.env.DBX_SQL_DEBUG;
|
||||
delete process.env.DBX_MCP_DEBUG_SQL;
|
||||
console.error = (...args: unknown[]) => messages.push(args);
|
||||
try {
|
||||
const server = createDbxMcpServer(backend, { isWebMode: true });
|
||||
const result = await (server as any)._registeredTools.dbx_execute_query.handler({
|
||||
connection_name: "local",
|
||||
sql: "select 'secret-123' as token",
|
||||
});
|
||||
assert.equal(result.isError, undefined);
|
||||
} finally {
|
||||
console.error = original;
|
||||
if (originalDebug === undefined) delete process.env.DBX_SQL_DEBUG;
|
||||
else process.env.DBX_SQL_DEBUG = originalDebug;
|
||||
if (originalMcpDebug === undefined) delete process.env.DBX_MCP_DEBUG_SQL;
|
||||
else process.env.DBX_MCP_DEBUG_SQL = originalMcpDebug;
|
||||
}
|
||||
|
||||
assert.equal(messages.length, 0);
|
||||
});
|
||||
|
||||
test("dbx_execute_query omits raw SQL from user-facing query errors", async () => {
|
||||
const sensitiveSql = "select 'secret-123' as token";
|
||||
const scopedBackend: Backend = {
|
||||
...backend,
|
||||
executeQuery: async () => {
|
||||
throw new Error("driver rejected statement");
|
||||
},
|
||||
};
|
||||
const server = createDbxMcpServer(scopedBackend, { isWebMode: true });
|
||||
|
||||
const result = await (server as any)._registeredTools.dbx_execute_query.handler({
|
||||
connection_name: "local",
|
||||
sql: sensitiveSql,
|
||||
});
|
||||
|
||||
assert.equal(result.isError, true);
|
||||
assert.match(result.content[0].text, /QUERY_ERROR: driver rejected statement/);
|
||||
assert.doesNotMatch(result.content[0].text, /secret-123|SQL:/);
|
||||
});
|
||||
|
||||
test("redis command tool blocks write commands in read-only MCP sessions", async () => {
|
||||
let executed = false;
|
||||
const redisConnection: ConnectionConfig = { ...connection, db_type: "redis" };
|
||||
|
|
@ -808,3 +854,66 @@ test("dbx_execute_query with connection_id routes correctly on bridge-backed (SS
|
|||
assert.equal(usedConfigs[0].host, "private.local");
|
||||
assert.equal(usedConfigs[0].ssh_enabled, true);
|
||||
});
|
||||
|
||||
// --- Dialect-aware `#` comment handling ---
|
||||
|
||||
test("dbx_execute_query splits PG `#` operator statements correctly", async () => {
|
||||
const executed: string[] = [];
|
||||
const scopedBackend: Backend = {
|
||||
...backend,
|
||||
executeQuery: async (_config, sql) => {
|
||||
executed.push(sql);
|
||||
return { columns: ["value"], rows: [{ value: 1 }], row_count: 1 };
|
||||
},
|
||||
};
|
||||
const server = createDbxMcpServer(scopedBackend, { isWebMode: true });
|
||||
|
||||
// On a postgres connection, `#` is an operator, not a comment.
|
||||
// `SELECT 1 # 2; SELECT 3` should produce TWO executeQuery calls.
|
||||
await (server as any)._registeredTools.dbx_execute_query.handler({
|
||||
connection_name: "local",
|
||||
sql: "SELECT 1 # 2; SELECT 3",
|
||||
});
|
||||
|
||||
assert.deepEqual(executed, ["SELECT 1 # 2", "SELECT 3"]);
|
||||
});
|
||||
|
||||
test("dbx_execute_query treats `#` as line comment on MySQL connections", async () => {
|
||||
const mysqlConn: ConnectionConfig = { ...connection, id: "mysql-1", name: "mysql-local", db_type: "mysql" };
|
||||
const executed: string[] = [];
|
||||
const scopedBackend: Backend = {
|
||||
...backend,
|
||||
loadConnections: async () => [mysqlConn],
|
||||
findConnection: async (name) => (name === "mysql-local" ? mysqlConn : undefined),
|
||||
executeQuery: async (_config, sql) => {
|
||||
executed.push(sql);
|
||||
return { columns: ["value"], rows: [{ value: 1 }], row_count: 1 };
|
||||
},
|
||||
};
|
||||
const server = createDbxMcpServer(scopedBackend, { isWebMode: true });
|
||||
|
||||
// On a mysql connection, `#` IS a line comment.
|
||||
// The `;` in `SELECT 1;` splits the first statement. The `# comment\nSELECT 2`
|
||||
// is a single statement — the `#` makes everything on that line a comment,
|
||||
// and after the newline `SELECT 2` continues (no `;` to split).
|
||||
await (server as any)._registeredTools.dbx_execute_query.handler({
|
||||
connection_name: "mysql-local",
|
||||
sql: "SELECT 1; # comment\nSELECT 2",
|
||||
});
|
||||
|
||||
assert.deepEqual(executed, ["SELECT 1", "# comment\nSELECT 2"]);
|
||||
});
|
||||
|
||||
test("dbx_execute_query blocks PG injection through `#` as comment in classification", async () => {
|
||||
// `SELECT 1 # 2; DELETE FROM t` on a postgres connection: the `#` is an operator,
|
||||
// so classification must see the DELETE and block it as a write in read-only mode.
|
||||
const server = createDbxMcpServer(backend, { isWebMode: true });
|
||||
|
||||
const result = await (server as any)._registeredTools.dbx_execute_query.handler({
|
||||
connection_name: "local",
|
||||
sql: "SELECT 1 # 2; DELETE FROM t",
|
||||
});
|
||||
|
||||
assert.equal(result.isError, true);
|
||||
assert.match(result.content[0].text, /SQL_BLOCKED:/);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@
|
|||
"./production-safety": "./dist/production-safety.js",
|
||||
"./redis-command": "./dist/redis-command.js",
|
||||
"./schema-context": "./dist/schema-context.js",
|
||||
"./sql-diagnostics": "./dist/sql-diagnostics.js",
|
||||
"./sql-risk": "./dist/sql-risk.js",
|
||||
"./sql-safety": "./dist/sql-safety.js"
|
||||
},
|
||||
|
|
|
|||
|
|
@ -970,7 +970,7 @@ async function executeRedisCommandDirect(config: ConnectionConfig, db: number, c
|
|||
const command = argv[0].toUpperCase();
|
||||
const safety = classifyRedisCommand(command) as RedisCommandSafety;
|
||||
if (!options?.skipSafetyCheck && safety === "blocked") {
|
||||
throw new Error(`Redis command is blocked for safety: ${command}`);
|
||||
throw new Error("Redis command is blocked for safety. Enable dangerous commands with DBX_MCP_ALLOW_DANGEROUS_SQL=1.");
|
||||
}
|
||||
|
||||
const { Redis } = await import("ioredis");
|
||||
|
|
|
|||
|
|
@ -9,5 +9,6 @@ export * from "./paths.js";
|
|||
export * from "./production-safety.js";
|
||||
export * from "./redis-command.js";
|
||||
export * from "./schema-context.js";
|
||||
export * from "./sql-diagnostics.js";
|
||||
export * from "./sql-risk.js";
|
||||
export * from "./sql-safety.js";
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import type { ConnectionConfig } from "./connections.js";
|
||||
import { classifySqlRisk, isSqlRiskMutation } from "./sql-risk.js";
|
||||
import { classifySqlRisk, isSqlRiskMutation, supportsHashLineComments } from "./sql-risk.js";
|
||||
|
||||
export interface ProductionSqlAssessment {
|
||||
active: boolean;
|
||||
|
|
@ -93,7 +93,8 @@ export function isProductionDatabase(config: ConnectionConfig | undefined, datab
|
|||
export function assessProductionSql(sql: string, config: ConnectionConfig | undefined, activeDatabase?: string): ProductionSqlAssessment {
|
||||
const targetText = sqlTargetSafetyText(sql);
|
||||
const statements = splitTargetStatements(targetText.text);
|
||||
const isMutation = isSqlRiskMutation(classifySqlRisk(sql).risk);
|
||||
const hashLineComments = supportsHashLineComments(config?.db_type);
|
||||
const isMutation = isSqlRiskMutation(classifySqlRisk(sql, { hashLineComments }).risk);
|
||||
if (!isMutation || !config) return { active: isProductionDatabase(config, activeDatabase), isMutation, databases: [] };
|
||||
if (config.is_production) return { active: true, isMutation, databases: [] };
|
||||
if (isProductionDatabase(config, activeDatabase)) return { active: true, isMutation, databases: activeDatabase ? [activeDatabase] : [] };
|
||||
|
|
@ -101,12 +102,12 @@ export function assessProductionSql(sql: string, config: ConnectionConfig | unde
|
|||
const marked = new Set((config.production_databases ?? []).map(normalizeProductionDatabase).filter(Boolean));
|
||||
if (!marked.size) return { active: false, isMutation, databases: [] };
|
||||
|
||||
const targets = referencedDatabases(statements, config.db_type, activeDatabase, targetText.quotedIdentifiers);
|
||||
const targets = referencedDatabases(statements, config.db_type, hashLineComments, activeDatabase, targetText.quotedIdentifiers);
|
||||
const databases = targets.databases.filter((database) => marked.has(normalizeProductionDatabase(database)));
|
||||
return { active: databases.length > 0 || targets.uncertain, isMutation, databases: databases.length > 0 ? databases : targets.uncertain ? [...marked] : [] };
|
||||
}
|
||||
|
||||
function referencedDatabases(statements: string[], dbType: string, activeDatabase: string | undefined, quotedIdentifiers: Map<string, string>): ReferencedDatabaseAssessment {
|
||||
function referencedDatabases(statements: string[], dbType: string, hashLineComments: boolean, activeDatabase: string | undefined, quotedIdentifiers: Map<string, string>): ReferencedDatabaseAssessment {
|
||||
const databases = new Set<string>();
|
||||
let uncertain = false;
|
||||
let useDatabase = "";
|
||||
|
|
@ -114,7 +115,7 @@ function referencedDatabases(statements: string[], dbType: string, activeDatabas
|
|||
|
||||
for (const statement of statements) {
|
||||
const statementDatabases = new Set<string>();
|
||||
const statementAssessment = classifySqlRisk(statement);
|
||||
const statementAssessment = classifySqlRisk(statement, { hashLineComments });
|
||||
const statementIsMutation = isSqlRiskMutation(statementAssessment.risk);
|
||||
const useMatch = statement.match(USE_RE);
|
||||
if (useMatch?.[1]) {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,130 @@
|
|||
const DEFAULT_SQL_DIAGNOSTIC_MAX_CHARS = 512;
|
||||
const SENSITIVE_NAME_RE = /(?:password|passwd|pwd|secret|token|api[_-]?key|access[_-]?key|private[_-]?key|credential|authorization|bearer)/i;
|
||||
|
||||
function boundedInput(sql: string, maxChars: number): [string, boolean] {
|
||||
if (maxChars <= 0) return ["", sql.length > 0];
|
||||
|
||||
let end = 0;
|
||||
let chars = 0;
|
||||
for (const character of sql) {
|
||||
if (chars === maxChars) return [sql.slice(0, end), true];
|
||||
end += character.length;
|
||||
chars += 1;
|
||||
}
|
||||
return [sql, false];
|
||||
}
|
||||
|
||||
function truncateForDiagnostic(value: string, maxChars: number, inputTruncated: boolean): string {
|
||||
if (value.length > maxChars) return `${value.slice(0, maxChars)}…[truncated]`;
|
||||
return inputTruncated ? `${value}…[truncated]` : value;
|
||||
}
|
||||
|
||||
function redactSqlLiterals(sql: string): string {
|
||||
let result = "";
|
||||
let i = 0;
|
||||
while (i < sql.length) {
|
||||
const ch = sql[i];
|
||||
const next = sql[i + 1];
|
||||
if (ch === "'" || ch === '"' || ch === "`") {
|
||||
const quote = ch;
|
||||
result += `${quote}[REDACTED]${quote}`;
|
||||
i += 1;
|
||||
while (i < sql.length) {
|
||||
const current = sql[i];
|
||||
if (current === quote) {
|
||||
if (sql[i + 1] === quote) {
|
||||
i += 2;
|
||||
continue;
|
||||
}
|
||||
i += 1;
|
||||
break;
|
||||
}
|
||||
if (current === "\\" && quote !== "`") {
|
||||
i += 2;
|
||||
} else {
|
||||
i += 1;
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (ch === "$") {
|
||||
const j = i + 1;
|
||||
if (j >= sql.length) {
|
||||
result += "$";
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
if (sql[j] === "$") {
|
||||
// $$...$$ empty-tag dollar-quoted string
|
||||
result += "$$[REDACTED]$$";
|
||||
i += 2;
|
||||
while (i + 1 < sql.length && !(sql[i] === "$" && sql[i + 1] === "$")) {
|
||||
i += 1;
|
||||
}
|
||||
if (i + 1 < sql.length) {
|
||||
i += 2;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
// $tag$...$tag$ dollar-quoted string — tag must be ASCII alphanumerics + underscore only
|
||||
const TAG_CHAR = /^[A-Za-z0-9_]$/;
|
||||
let tagEnd = j;
|
||||
while (tagEnd < sql.length && TAG_CHAR.test(sql[tagEnd])) {
|
||||
tagEnd += 1;
|
||||
}
|
||||
if (tagEnd > j && tagEnd < sql.length && sql[tagEnd] === "$") {
|
||||
const tag = sql.slice(j, tagEnd);
|
||||
result += "$[REDACTED]$";
|
||||
i = tagEnd + 1;
|
||||
const closing = "$" + tag + "$";
|
||||
while (i + closing.length <= sql.length) {
|
||||
if (sql.slice(i, i + closing.length) === closing) {
|
||||
i += closing.length;
|
||||
break;
|
||||
}
|
||||
i += 1;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
result += "$";
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
if (ch === "-" && next === "-") {
|
||||
result += "--[REDACTED_COMMENT]";
|
||||
i += 2;
|
||||
while (i < sql.length && sql[i] !== "\n" && sql[i] !== "\r") i += 1;
|
||||
continue;
|
||||
}
|
||||
if (ch === "/" && next === "*") {
|
||||
result += "/*[REDACTED_COMMENT]*/";
|
||||
i += 2;
|
||||
while (i < sql.length && !(sql[i] === "*" && sql[i + 1] === "/")) i += 1;
|
||||
if (i < sql.length) i += 2;
|
||||
continue;
|
||||
}
|
||||
result += ch;
|
||||
i += 1;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
export function redactSqlForDiagnostics(sql: string, maxChars = DEFAULT_SQL_DIAGNOSTIC_MAX_CHARS): string {
|
||||
const [boundedSql, inputTruncated] = boundedInput(sql, maxChars);
|
||||
const literalRedacted = redactSqlLiterals(boundedSql);
|
||||
const sensitiveRedacted = literalRedacted.replace(/\b([A-Za-z_][\w.-]*)(\s*[:=]\s*)([^\s,;)]+)/g, (match, key: string, separator: string) => {
|
||||
if (!SENSITIVE_NAME_RE.test(key)) return match;
|
||||
return `${key}${separator}[REDACTED]`;
|
||||
});
|
||||
return truncateForDiagnostic(sensitiveRedacted, maxChars, inputTruncated);
|
||||
}
|
||||
|
||||
export function sqlDiagnosticsEnabled(env: NodeJS.ProcessEnv = process.env): boolean {
|
||||
const value = env.DBX_SQL_DEBUG ?? env.DBX_DEBUG_SQL ?? env.DBX_MCP_DEBUG_SQL;
|
||||
return value === "1" || value?.toLowerCase() === "true";
|
||||
}
|
||||
|
||||
export function logSqlDiagnostic(scope: string, sql: string, details: Record<string, unknown> = {}, env?: NodeJS.ProcessEnv): void {
|
||||
if (!sqlDiagnosticsEnabled(env)) return;
|
||||
console.error(`[${scope}] sql:`, JSON.stringify({ ...details, sql: redactSqlForDiagnostics(sql) }));
|
||||
}
|
||||
|
|
@ -9,6 +9,25 @@ export interface SqlRiskAssessment extends SqlRiskStatementAssessment {
|
|||
statements: SqlRiskStatementAssessment[];
|
||||
}
|
||||
|
||||
/** Options for SQL text utilities that parse comments and literals. */
|
||||
export interface SqlTextOptions {
|
||||
/** Whether `#` starts a line comment. Only MySQL-family databases support this.
|
||||
* Default: false (fail-safe — `#` is treated as an operator, which may over-block
|
||||
* MySQL classification but never under-blocks PostgreSQL). */
|
||||
hashLineComments?: boolean;
|
||||
}
|
||||
|
||||
/** Database types whose SQL dialect uses `#` for line comments (MySQL family). */
|
||||
const MYSQL_HASH_COMMENT_DB_TYPES = new Set(["mysql", "doris", "starrocks", "manticoresearch", "goldendb"]);
|
||||
|
||||
/** Determine whether the given database type supports `#` line comments.
|
||||
* Mirrors the Rust `is_mysql_compatible_database` dialect set:
|
||||
* Mysql, Doris, StarRocks, ManticoreSearch, Goldendb. */
|
||||
export function supportsHashLineComments(dbType?: string): boolean {
|
||||
if (!dbType) return false;
|
||||
return MYSQL_HASH_COMMENT_DB_TYPES.has(dbType);
|
||||
}
|
||||
|
||||
interface SqlRiskToken {
|
||||
text: string;
|
||||
normalized: string;
|
||||
|
|
@ -23,15 +42,15 @@ const PRIMARY_STATEMENT_KEYWORDS = new Set([...READ_KEYWORDS, ...WRITE_KEYWORDS,
|
|||
const SAFE_READ_PRAGMA_NAMES = new Set(["table_info", "table_xinfo", "index_list", "index_info", "foreign_key_list", "database_list", "compile_options", "data_version"]);
|
||||
const RISK_ORDER: Record<SqlRiskLevel, number> = { read: 0, write: 1, ddl: 2, transaction: 3, unknown: 4 };
|
||||
|
||||
export function splitSqlStatementsForSafety(sql: string): string[] {
|
||||
return sqlSafetyText(sql)
|
||||
export function splitSqlStatementsForSafety(sql: string, options?: SqlTextOptions): string[] {
|
||||
return sqlSafetyText(sql, options)
|
||||
.split(";")
|
||||
.map((statement) => statement.trim())
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
export function classifySqlRisk(sql: string): SqlRiskAssessment {
|
||||
const statements = splitSqlStatementsForSafety(sql).map(classifySqlStatementRisk);
|
||||
export function classifySqlRisk(sql: string, options?: SqlTextOptions): SqlRiskAssessment {
|
||||
const statements = splitSqlStatementsForSafety(sql, options).map(classifySqlStatementRisk);
|
||||
if (!statements.length) return { risk: "unknown", statements: [] };
|
||||
const highest = statements.reduce<SqlRiskStatementAssessment>((current, statement) => (RISK_ORDER[statement.risk] > RISK_ORDER[current.risk] ? statement : current), { risk: "read" });
|
||||
return { ...highest, statements };
|
||||
|
|
@ -45,9 +64,10 @@ export function isSqlRiskMutation(risk: SqlRiskLevel): boolean {
|
|||
return risk !== "read";
|
||||
}
|
||||
|
||||
export function sqlSafetyText(sql: string): string {
|
||||
export function sqlSafetyText(sql: string, options?: SqlTextOptions): string {
|
||||
let output = "";
|
||||
let index = 0;
|
||||
const hashLineComments = options?.hashLineComments === true;
|
||||
while (index < sql.length) {
|
||||
const char = sql[index] ?? "";
|
||||
const next = sql[index + 1] ?? "";
|
||||
|
|
@ -57,7 +77,7 @@ export function sqlSafetyText(sql: string): string {
|
|||
output += " ";
|
||||
continue;
|
||||
}
|
||||
if (char === "#") {
|
||||
if (hashLineComments && char === "#") {
|
||||
index += 1;
|
||||
while (index < sql.length && sql[index] !== "\n" && sql[index] !== "\r") index += 1;
|
||||
output += " ";
|
||||
|
|
@ -69,7 +89,7 @@ export function sqlSafetyText(sql: string): string {
|
|||
const executablePrefixLength = mysqlExecutableCommentPrefixLength(sql, index);
|
||||
if (executablePrefixLength > 0) {
|
||||
const bodyStart = skipExecutableCommentVersion(sql, index + executablePrefixLength);
|
||||
output += ` ${sqlSafetyText(sql.slice(bodyStart, close))} `;
|
||||
output += ` ${sqlSafetyText(sql.slice(bodyStart, close), options)} `;
|
||||
} else {
|
||||
output += " ";
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,9 +1,11 @@
|
|||
import { classifySqlStatementRisk, splitSqlStatementsForSafety, sqlSafetyText } from "./sql-risk.js";
|
||||
import { classifySqlStatementRisk, splitSqlStatementsForSafety, sqlSafetyText, type SqlTextOptions } from "./sql-risk.js";
|
||||
|
||||
export interface SqlSafetyOptions {
|
||||
allowWrites?: boolean;
|
||||
allowDangerous?: boolean;
|
||||
allowMultipleStatements?: boolean;
|
||||
/** Whether `#` starts a line comment (MySQL family only). Default: false. */
|
||||
hashLineComments?: boolean;
|
||||
}
|
||||
|
||||
export interface SqlSafetyDecision {
|
||||
|
|
@ -22,7 +24,7 @@ function parseBooleanEnv(value: string | undefined): boolean | undefined {
|
|||
}
|
||||
|
||||
export function evaluateSqlSafety(sql: string, options: SqlSafetyOptions = {}): SqlSafetyDecision {
|
||||
const statements = splitSqlStatementsForSafety(sql);
|
||||
const statements = splitSqlStatementsForSafety(sql, options);
|
||||
if (statements.length === 0) return { allowed: false, reason: "SQL is empty." };
|
||||
if (statements.length > 1 && !options.allowMultipleStatements) {
|
||||
return { allowed: false, reason: "Only one SQL statement is allowed per query." };
|
||||
|
|
@ -59,7 +61,7 @@ function evaluateSingleSqlStatementSafety(sql: string, options: SqlSafetyOptions
|
|||
}
|
||||
|
||||
if (options.allowWrites && !options.allowDangerous) {
|
||||
const tokens: string[] = sqlSafetyText(sql).toLowerCase().match(/[a-z_]+/g) ?? [];
|
||||
const tokens: string[] = sqlSafetyText(sql, options).toLowerCase().match(/[a-z_]+/g) ?? [];
|
||||
if (firstKeyword === "update" && !tokens.includes("where")) {
|
||||
return { allowed: false, reason: "UPDATE statements must include a WHERE clause." };
|
||||
}
|
||||
|
|
@ -80,12 +82,13 @@ export function sqlSafetyFromEnv(env: NodeJS.ProcessEnv = process.env): SqlSafet
|
|||
};
|
||||
}
|
||||
|
||||
export function splitSqlStatements(sql: string): string[] {
|
||||
export function splitSqlStatements(sql: string, options?: SqlTextOptions): string[] {
|
||||
const statements: string[] = [];
|
||||
let statementStart = 0;
|
||||
let index = 0;
|
||||
let state: "none" | "single" | "double" | "backtick" | "bracket" | "lineComment" | "blockComment" | "dollar" = "none";
|
||||
let dollarTag = "";
|
||||
const hashLineComments = options?.hashLineComments === true;
|
||||
|
||||
const pushStatement = (end: number) => {
|
||||
const statement = sql.slice(statementStart, end).trim();
|
||||
|
|
@ -152,7 +155,7 @@ export function splitSqlStatements(sql: string): string[] {
|
|||
index += 2;
|
||||
continue;
|
||||
}
|
||||
if (char === "#") {
|
||||
if (hashLineComments && char === "#") {
|
||||
state = "lineComment";
|
||||
index += 1;
|
||||
continue;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,107 @@
|
|||
import assert from "node:assert/strict";
|
||||
import { test } from "vitest";
|
||||
import { logSqlDiagnostic, redactSqlForDiagnostics, sqlDiagnosticsEnabled } from "../src/sql-diagnostics.js";
|
||||
|
||||
test("SQL diagnostics are disabled unless explicitly enabled", () => {
|
||||
assert.equal(sqlDiagnosticsEnabled({}), false);
|
||||
assert.equal(sqlDiagnosticsEnabled({ DBX_SQL_DEBUG: "0" }), false);
|
||||
assert.equal(sqlDiagnosticsEnabled({ DBX_SQL_DEBUG: "1" }), true);
|
||||
assert.equal(sqlDiagnosticsEnabled({ DBX_MCP_DEBUG_SQL: "true" }), true);
|
||||
});
|
||||
|
||||
test("redacts sensitive literals and bounds large SQL diagnostics", () => {
|
||||
const sql = `select * from users where password = 'secret-123' and token="tok-456" and api_key=plain ${"x".repeat(900)}`;
|
||||
const redacted = redactSqlForDiagnostics(sql);
|
||||
|
||||
assert.doesNotMatch(redacted, /secret-123|tok-456|api_key=plain/);
|
||||
assert.match(redacted, /\[REDACTED\]/);
|
||||
assert.match(redacted, /api_key=\[REDACTED\]/);
|
||||
assert.match(redacted, /truncated/);
|
||||
assert.ok(redacted.length < sql.length);
|
||||
});
|
||||
|
||||
test("disabled SQL diagnostic logging does not write statements", () => {
|
||||
const original = console.error;
|
||||
const messages: unknown[][] = [];
|
||||
console.error = (...args: unknown[]) => messages.push(args);
|
||||
try {
|
||||
logSqlDiagnostic("test", "select 'secret-123'", {}, {});
|
||||
} finally {
|
||||
console.error = original;
|
||||
}
|
||||
|
||||
assert.equal(messages.length, 0);
|
||||
});
|
||||
|
||||
test("enabled SQL diagnostic logging emits redacted statements only", () => {
|
||||
const original = console.error;
|
||||
const messages: unknown[][] = [];
|
||||
console.error = (...args: unknown[]) => messages.push(args);
|
||||
try {
|
||||
logSqlDiagnostic("test", "select 'secret-123' as password", {}, { DBX_SQL_DEBUG: "1" });
|
||||
} finally {
|
||||
console.error = original;
|
||||
}
|
||||
|
||||
assert.equal(messages.length, 1);
|
||||
const rendered = messages.flat().join(" ");
|
||||
assert.doesNotMatch(rendered, /secret-123/);
|
||||
assert.match(rendered, /\[REDACTED\]/);
|
||||
});
|
||||
|
||||
test("dollar-quoted strings are redacted", () => {
|
||||
const redacted = redactSqlForDiagnostics("select $$secret$$");
|
||||
assert.doesNotMatch(redacted, /secret/);
|
||||
assert.match(redacted, /\[REDACTED\]/);
|
||||
});
|
||||
|
||||
test("space-separated sensitive assignments are redacted", () => {
|
||||
const redacted = redactSqlForDiagnostics("select * from t where password = mysecret");
|
||||
assert.doesNotMatch(redacted, /mysecret/);
|
||||
assert.match(redacted, /\[REDACTED\]/);
|
||||
});
|
||||
|
||||
test("postgres positional parameters are not treated as dollar quotes ($1, $2, ...)", () => {
|
||||
const redacted = redactSqlForDiagnostics("select * from t where id = $1 and name = 'alice'");
|
||||
assert.match(redacted, /\$1\b/);
|
||||
assert.doesNotMatch(redacted, /alice/);
|
||||
});
|
||||
|
||||
test("multiple postgres positional parameters all survive redaction", () => {
|
||||
const redacted = redactSqlForDiagnostics("select $1, $2, $3, $42 from t");
|
||||
assert.match(redacted, /\$1\b/);
|
||||
assert.match(redacted, /\$2\b/);
|
||||
assert.match(redacted, /\$3\b/);
|
||||
assert.match(redacted, /\$42\b/);
|
||||
});
|
||||
|
||||
test("empty-tag dollar quote $$secret$$ is redacted", () => {
|
||||
const redacted = redactSqlForDiagnostics("select $$secret$$ from t");
|
||||
assert.match(redacted, /\$\$\[REDACTED\]\$\$/);
|
||||
assert.doesNotMatch(redacted, /secret/);
|
||||
});
|
||||
|
||||
test("named-tag dollar quote $tag$hello$tag$ is redacted", () => {
|
||||
const redacted = redactSqlForDiagnostics("select $tag$hello$tag$ from t");
|
||||
assert.match(redacted, /\$\[REDACTED\]\$/);
|
||||
assert.doesNotMatch(redacted, /hello/);
|
||||
});
|
||||
|
||||
test("lone trailing dollar sign does not throw and passes through", () => {
|
||||
const redacted = redactSqlForDiagnostics("select 1 $");
|
||||
assert.match(redacted, /\$/);
|
||||
});
|
||||
|
||||
test("bounds redaction before scanning an unclosed literal", () => {
|
||||
const redacted = redactSqlForDiagnostics(`select '${"secret-".repeat(1000)}`, 32);
|
||||
assert.doesNotMatch(redacted, /secret-/);
|
||||
assert.match(redacted, /\[REDACTED\]/);
|
||||
assert.match(redacted, /truncated/);
|
||||
});
|
||||
|
||||
test("does not leak a sensitive value cut at the diagnostic boundary", () => {
|
||||
const redacted = redactSqlForDiagnostics(`password = ${"secret-token".repeat(1000)}`, 24);
|
||||
assert.doesNotMatch(redacted, /secret-token|secret-/);
|
||||
assert.match(redacted, /password = \[REDACTED\]/);
|
||||
assert.match(redacted, /truncated/);
|
||||
});
|
||||
|
|
@ -1,6 +1,7 @@
|
|||
import assert from "node:assert/strict";
|
||||
import { test } from "vitest";
|
||||
import { evaluateSqlSafety, splitSqlStatements, sqlSafetyFromEnv } from "../src/sql-safety.js";
|
||||
import { supportsHashLineComments } from "../src/sql-risk.js";
|
||||
|
||||
test("allows read-only SQL by default", () => {
|
||||
const decision = evaluateSqlSafety("select * from users limit 5");
|
||||
|
|
@ -110,3 +111,77 @@ test("sqlSafetyFromEnv supports explicitly disabling writes", () => {
|
|||
assert.equal(options.allowWrites, false);
|
||||
assert.equal(options.allowDangerous, false);
|
||||
});
|
||||
|
||||
// --- Dialect-aware `#` comment handling ---
|
||||
|
||||
test("supportsHashLineComments matches Rust mysql-compatible dialect set", () => {
|
||||
for (const dbType of ["mysql", "doris", "starrocks", "manticoresearch", "goldendb"]) {
|
||||
assert.equal(supportsHashLineComments(dbType), true, dbType);
|
||||
}
|
||||
for (const dbType of ["postgres", "sqlite", "sqlserver", "oracle", "duckdb", "bigquery", "redshift", ""]) {
|
||||
assert.equal(supportsHashLineComments(dbType), false, dbType);
|
||||
}
|
||||
assert.equal(supportsHashLineComments(undefined), false);
|
||||
});
|
||||
|
||||
test("splitSqlStatements splits PG `#` operator correctly (hashLineComments omitted/default)", () => {
|
||||
assert.deepEqual(splitSqlStatements("SELECT a # b; SELECT 2"), ["SELECT a # b", "SELECT 2"]);
|
||||
});
|
||||
|
||||
test("splitSqlStatements splits PG `#` operator correctly (hashLineComments: false)", () => {
|
||||
assert.deepEqual(splitSqlStatements("SELECT a # b; SELECT 2", { hashLineComments: false }), [
|
||||
"SELECT a # b",
|
||||
"SELECT 2",
|
||||
]);
|
||||
});
|
||||
|
||||
test("splitSqlStatements treats `#` as comment with hashLineComments: true (MySQL)", () => {
|
||||
// With hashLineComments: true, the `;` inside the `#` comment must NOT split.
|
||||
// The comment text is preserved in the output (splitter only delimits on `;`, it doesn't strip).
|
||||
assert.deepEqual(
|
||||
splitSqlStatements("SELECT 1; # trailing ; comment\nSELECT 2", { hashLineComments: true }),
|
||||
["SELECT 1", "# trailing ; comment\nSELECT 2"],
|
||||
);
|
||||
});
|
||||
|
||||
test("splitSqlStatements preserves JSONB operator text verbatim", () => {
|
||||
const result = splitSqlStatements("SELECT data #>> '{a,b}' FROM t");
|
||||
assert.equal(result.length, 1);
|
||||
assert.equal(result[0], "SELECT data #>> '{a,b}' FROM t");
|
||||
});
|
||||
|
||||
test("splitSqlStatements handles `#` as operator mid-statement (PG)", () => {
|
||||
assert.deepEqual(splitSqlStatements("SELECT 1 # 2; DELETE FROM t"), [
|
||||
"SELECT 1 # 2",
|
||||
"DELETE FROM t",
|
||||
]);
|
||||
});
|
||||
|
||||
test("evaluateSqlSafety blocks PG injection that bypasses # as comment (regression)", () => {
|
||||
// Before fix: # would strip "2; DELETE FROM t" as comment, classify as read-only.
|
||||
// After fix: # is treated as an operator, so DELETE FROM t is seen as a second write statement.
|
||||
const decision = evaluateSqlSafety("SELECT 1 # 2; DELETE FROM t", {
|
||||
allowWrites: false,
|
||||
allowMultipleStatements: true,
|
||||
});
|
||||
assert.equal(decision.allowed, false);
|
||||
assert.match(decision.reason ?? "", /read-only/i);
|
||||
});
|
||||
|
||||
test("evaluateSqlSafety allows MySQL `#` comment with hashLineComments: true", () => {
|
||||
const decision = evaluateSqlSafety("SELECT 1 # delete note", {
|
||||
allowWrites: false,
|
||||
allowMultipleStatements: true,
|
||||
hashLineComments: true,
|
||||
});
|
||||
assert.equal(decision.allowed, true);
|
||||
});
|
||||
|
||||
test("evaluateSqlSafety with hashLineComments: false still sees DELETE after `#` operator", () => {
|
||||
const decision = evaluateSqlSafety("SELECT 1 # 2; DELETE FROM t", {
|
||||
allowWrites: false,
|
||||
allowMultipleStatements: true,
|
||||
hashLineComments: false,
|
||||
});
|
||||
assert.equal(decision.allowed, false);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -85,13 +85,13 @@ pub async fn execute_multi(
|
|||
let cancel_token = registered_query.as_ref().map(|query| query.token());
|
||||
let trace_id = execution_id.as_deref().unwrap_or("no-execution-id").to_string();
|
||||
let started_at = Instant::now();
|
||||
dbx_core::sql_diagnostics::debug_sql("query:execute_multi:start", &sql);
|
||||
log::info!(
|
||||
"[query][execute_multi:start] trace_id={} connection_id={} database={} schema={:?} sql={}",
|
||||
"[query][execute_multi:start] trace_id={} connection_id={} database={} schema={:?}",
|
||||
trace_id,
|
||||
connection_id,
|
||||
database,
|
||||
schema,
|
||||
sql
|
||||
schema
|
||||
);
|
||||
|
||||
let result = dbx_core::query::execute_multi_core_with_options_for_client(
|
||||
|
|
|
|||
Loading…
Reference in New Issue