use std::time::{Duration, Instant};
use axum::Json;
use axum::extract::State;
use axum::http::StatusCode;
use futures::TryStreamExt;
use serde::{Deserialize, Serialize};
use sqlx::{AssertSqlSafe, Column, Executor, Row, SqlSafeStr, SqlitePool, TypeInfo, ValueRef};
use tracing::warn;
use super::AppState;
const DEFAULT_LIMIT: usize = 1_000;
const MAX_LIMIT: usize = 10_000;
const QUERY_TIMEOUT: Duration = Duration::from_secs(15);
#[derive(Deserialize)]
pub struct QueryRequest {
pub sql: String,
#[serde(default)]
pub limit: Option<usize>,
}
#[derive(Serialize)]
pub struct QueryResponse {
pub columns: Vec<String>,
pub rows: Vec<Vec<serde_json::Value>>,
pub row_count: usize,
pub truncated: bool,
pub elapsed_ms: u128,
}
pub fn validate_read_only(sql: &str) -> Result<(), String> {
let trimmed = sql.trim();
if trimmed.is_empty() {
return Err("empty query".into());
}
let kw_end = trimmed
.find(|c: char| !c.is_ascii_alphabetic())
.unwrap_or(trimmed.len());
let kw = trimmed[..kw_end].to_ascii_lowercase();
if kw != "select" && kw != "with" {
return Err(format!(
"only read-only SELECT / WITH queries are allowed (got '{kw}')"
));
}
let stripped = strip_sql_noise(trimmed).to_ascii_lowercase();
if stripped.trim_end_matches(';').contains(';') {
return Err("multiple statements are not allowed".into());
}
for kw in [
"insert", "update", "delete", "replace", "drop", "create", "alter", "attach", "detach",
"pragma", "vacuum", "reindex",
] {
if contains_word(&stripped, kw) {
return Err(format!("'{kw}' is not allowed in a read-only query"));
}
}
Ok(())
}
fn strip_sql_noise(sql: &str) -> String {
let mut out = String::with_capacity(sql.len());
let mut chars = sql.chars().peekable();
while let Some(c) = chars.next() {
match c {
'\'' | '"' => {
let quote = c;
while let Some(d) = chars.next() {
if d == quote {
if chars.peek() == Some("e) {
chars.next();
} else {
break;
}
}
}
out.push(' ');
}
'-' if chars.peek() == Some(&'-') => {
for d in chars.by_ref() {
if d == '\n' {
out.push('\n');
break;
}
}
}
'/' if chars.peek() == Some(&'*') => {
chars.next(); let mut prev = ' ';
for d in chars.by_ref() {
if prev == '*' && d == '/' {
break;
}
prev = d;
}
out.push(' ');
}
_ => out.push(c),
}
}
out
}
fn contains_word(haystack: &str, needle: &str) -> bool {
let is_word = |c: char| c.is_alphanumeric() || c == '_';
let mut start = 0;
while let Some(pos) = haystack[start..].find(needle) {
let i = start + pos;
let before_ok = haystack[..i]
.chars()
.next_back()
.is_none_or(|c| !is_word(c));
let after = i + needle.len();
let after_ok = haystack[after..].chars().next().is_none_or(|c| !is_word(c));
if before_ok && after_ok {
return true;
}
start = after;
}
false
}
fn decode_cell(row: &sqlx::sqlite::SqliteRow, i: usize) -> serde_json::Value {
use serde_json::Value;
let class = match row.try_get_raw(i) {
Err(_) => return Value::Null,
Ok(raw) if raw.is_null() => return Value::Null,
Ok(raw) => raw.type_info().name().to_string(),
};
match class.as_str() {
"INTEGER" => row
.try_get::<i64, _>(i)
.map(Value::from)
.unwrap_or(Value::Null),
"REAL" => row
.try_get::<f64, _>(i)
.ok()
.and_then(serde_json::Number::from_f64)
.map(Value::Number)
.unwrap_or(Value::Null),
"BLOB" => row
.try_get::<Vec<u8>, _>(i)
.map(|b| Value::String(format!("<blob {} bytes>", b.len())))
.unwrap_or(Value::Null),
_ => row
.try_get::<String, _>(i)
.map(Value::String)
.unwrap_or(Value::Null),
}
}
pub(crate) struct ReadOnlyResult {
pub columns: Vec<String>,
pub rows: Vec<Vec<serde_json::Value>>,
pub truncated: bool,
}
pub(crate) enum ReadOnlyError {
Rejected(String),
Timeout,
Sql(String),
}
impl std::fmt::Display for ReadOnlyError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ReadOnlyError::Rejected(m) => write!(f, "{m}"),
ReadOnlyError::Timeout => {
write!(f, "query exceeded the {}s budget", QUERY_TIMEOUT.as_secs())
}
ReadOnlyError::Sql(m) => write!(f, "query failed: {m}"),
}
}
}
pub(crate) async fn run_read_only(
pool: &SqlitePool,
sql: &str,
limit: usize,
) -> Result<ReadOnlyResult, ReadOnlyError> {
validate_read_only(sql).map_err(ReadOnlyError::Rejected)?;
let limit = limit.clamp(1, MAX_LIMIT);
let fetch = async {
let described = pool
.describe(AssertSqlSafe(sql.to_string()).into_sql_str())
.await?;
let columns: Vec<String> = described
.columns()
.iter()
.map(|c| c.name().to_string())
.collect();
let mut stream = sqlx::query(AssertSqlSafe(sql.to_string())).fetch(pool);
let mut rows: Vec<Vec<serde_json::Value>> = Vec::new();
let mut truncated = false;
while let Some(row) = stream.try_next().await? {
if rows.len() >= limit {
truncated = true;
break;
}
let cells = (0..row.columns().len())
.map(|i| decode_cell(&row, i))
.collect();
rows.push(cells);
}
Ok::<_, sqlx::Error>(ReadOnlyResult {
columns,
rows,
truncated,
})
};
match tokio::time::timeout(QUERY_TIMEOUT, fetch).await {
Err(_) => Err(ReadOnlyError::Timeout),
Ok(Err(e)) => {
warn!(error = %e, "read-only query failed");
Err(ReadOnlyError::Sql(e.to_string()))
}
Ok(Ok(v)) => Ok(v),
}
}
pub async fn execute(
State(state): State<AppState>,
Json(req): Json<QueryRequest>,
) -> Result<Json<QueryResponse>, (StatusCode, String)> {
let started = Instant::now();
let limit = req.limit.unwrap_or(DEFAULT_LIMIT);
let result = run_read_only(&state.query_pool, &req.sql, limit)
.await
.map_err(|e| {
let status = match e {
ReadOnlyError::Rejected(_) => StatusCode::BAD_REQUEST,
ReadOnlyError::Timeout => StatusCode::REQUEST_TIMEOUT,
ReadOnlyError::Sql(_) => StatusCode::BAD_REQUEST,
};
(status, e.to_string())
})?;
let row_count = result.rows.len();
Ok(Json(QueryResponse {
columns: result.columns,
rows: result.rows,
row_count,
truncated: result.truncated,
elapsed_ms: started.elapsed().as_millis(),
}))
}
#[cfg(test)]
mod tests {
use super::*;
use sqlx::sqlite::SqlitePoolOptions;
#[test]
fn accepts_select_and_with() {
validate_read_only("SELECT 1").unwrap();
validate_read_only(" select * from t ").unwrap();
validate_read_only("WITH x AS (SELECT 1) SELECT * FROM x").unwrap();
validate_read_only("SELECT 1;").unwrap(); validate_read_only("SELECT * FROM t WHERE notes LIKE '%;%'").unwrap();
validate_read_only("SELECT * FROM audit WHERE action = 'delete'").unwrap();
validate_read_only("SELECT \"delete\" FROM t").unwrap();
validate_read_only("SELECT created_at, updated_at FROM t -- attach\n").unwrap();
}
#[test]
fn rejects_writes_and_ddl() {
for sql in [
"INSERT INTO t VALUES (1)",
"UPDATE t SET a = 1",
"DELETE FROM t",
"DROP TABLE t",
"CREATE TABLE t (a)",
"REPLACE INTO t VALUES (1)",
"",
" ",
] {
assert!(validate_read_only(sql).is_err(), "should reject: {sql:?}");
}
}
#[test]
fn rejects_stacked_statements() {
assert!(validate_read_only("SELECT 1; DROP TABLE t").is_err());
assert!(validate_read_only("SELECT 1; SELECT 2").is_err());
}
#[test]
fn rejects_attach_detach_pragma_and_write_verbs() {
assert!(validate_read_only("ATTACH DATABASE 'x' AS y").is_err());
assert!(validate_read_only("SELECT * FROM t PRAGMA foo").is_err());
assert!(validate_read_only("VACUUM").is_err());
assert!(validate_read_only("WITH x AS (SELECT 1) DELETE FROM t").is_err());
validate_read_only("SELECT cattached FROM pragmatic_table").unwrap();
validate_read_only("SELECT created_at, updated_at FROM deleted_log").unwrap();
}
#[test]
fn contains_word_is_boundary_aware() {
assert!(contains_word("select 1; attach 'x'", "attach"));
assert!(!contains_word("select cattached from t", "attach"));
assert!(!contains_word("select pragmatic from t", "pragma"));
assert!(!contains_word("cafédelete", "delete"));
}
#[test]
fn strip_sql_noise_removes_strings_idents_comments() {
assert_eq!(strip_sql_noise("a 'b;c' d"), "a d");
assert_eq!(strip_sql_noise("a \"x;y\" d"), "a d");
assert_eq!(strip_sql_noise("a -- b;c\nd"), "a \nd");
assert_eq!(strip_sql_noise("a /* b;c */ d"), "a d");
assert_eq!(strip_sql_noise("'it''s' x"), " x");
}
#[tokio::test]
async fn decode_cell_covers_storage_classes() {
let pool = SqlitePoolOptions::new()
.connect("sqlite::memory:")
.await
.unwrap();
let row = sqlx::query("SELECT 42 AS i, 1.5 AS r, 'hi' AS t, NULL AS n")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(decode_cell(&row, 0), serde_json::json!(42));
assert_eq!(decode_cell(&row, 1), serde_json::json!(1.5));
assert_eq!(decode_cell(&row, 2), serde_json::json!("hi"));
assert_eq!(decode_cell(&row, 3), serde_json::Value::Null);
}
}