SkardiLabs · BtXin · Jul 23, 2026 · Jul 17, 2026 · Jul 17, 2026 · Jul 17, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -64,7 +64,6 @@ rstest = "0.25.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_yaml = "0.9"
-sqlparser = "0.55.0"
 sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "tls-rustls", "postgres"] }
 tempfile = "3.23.0"
 tokio = { version = "1.44.2", features = ["macros", "rt", "sync"] }

diff --git a/crates/server/src/auth/bridge.rs b/crates/server/src/auth/bridge.rs
@@ -178,8 +178,15 @@ impl TableProvider for AuthSessionsTable {
 // Registration helper
 // ---------------------------------------------------------------------------
 
+/// Schema the auth virtual tables (`users`, `sessions`) register under.
+///
+/// Single source of truth for the reserved schema name: the ad-hoc `/query`
+/// policy denies exactly this schema (see `config::adhoc_policy_from_sources`),
+/// so the two must never drift apart.
+pub const AUTH_SCHEMA: &str = "auth";
+
 /// Register `auth.users` and `auth.sessions` virtual tables into the given
-/// DataFusion `SessionContext` under a dedicated `auth` schema.
+/// DataFusion `SessionContext` under the [`AUTH_SCHEMA`] schema.
 pub fn register_auth_tables(
     ctx: &mut SessionContext,
     auth: Arc<BetterAuth<DieselSqliteAdapter>>,
@@ -204,12 +211,14 @@ pub fn register_auth_tables(
         .catalog("datafusion")
         .ok_or_else(|| anyhow::anyhow!("Default catalog 'datafusion' not found"))?;
 
-    if catalog.schema("auth").is_some() {
-        return Err(anyhow::anyhow!("Auth schema 'auth' is already registered"));
+    if catalog.schema(AUTH_SCHEMA).is_some() {
+        return Err(anyhow::anyhow!(
+            "Auth schema '{AUTH_SCHEMA}' is already registered"
+        ));
     }
 
     catalog
-        .register_schema("auth", schema)
+        .register_schema(AUTH_SCHEMA, schema)
         .map_err(|e| anyhow::anyhow!("Failed to register auth schema: {}", e))?;
 
     tracing::info!("Registered DataFusion tables: auth.users, auth.sessions");

diff --git a/crates/server/src/auth/routes.rs b/crates/server/src/auth/routes.rs
@@ -8,6 +8,7 @@
 use std::collections::HashMap;
 
 use axum::{
+    Json,
     body::Body,
     extract::State,
     http::{HeaderName, HeaderValue, Request, Response, StatusCode},
@@ -16,6 +17,7 @@ use axum::{
 use better_auth::{AuthRequest, HttpMethod, SessionOps};
 use cookie::Cookie;
 
+use crate::response::{ErrorResponse, create_error_response};
 use crate::server::AppState;
 
 async fn to_auth_request(req: Request<Body>) -> Result<AuthRequest, String> {
@@ -181,6 +183,27 @@ pub async fn verify_session(
     }
 }
 
+/// Enforce the session gate for JSON API handlers: on auth failure, convert
+/// the raw `verify_session` response into the shared `ErrorResponse`
+/// envelope handlers return.
+pub(crate) async fn require_session(
+    state: &AppState,
+    headers: &axum::http::HeaderMap,
+) -> Result<(), (StatusCode, Json<ErrorResponse>)> {
+    if let Err(unauth_response) = verify_session(state, headers).await {
+        let status = unauth_response.status();
+        let body_bytes = axum::body::to_bytes(unauth_response.into_body(), 512)
+            .await
+            .unwrap_or_default();
+        let msg = serde_json::from_slice::<serde_json::Value>(&body_bytes)
+            .ok()
+            .and_then(|v| v["error"].as_str().map(|s| s.to_string()))
+            .unwrap_or_else(|| "Authentication required".to_string());
+        return Err((status, create_error_response(&msg, "unauthorized", None)));
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -340,13 +363,12 @@ mod tests {
     fn make_no_auth_state() -> AppState {
         use crate::auth::layer::AuthLayer;
         use crate::config::{CliArgs, ServerConfig};
-        use crate::metrics::PipelineMetrics;
         use crate::semantics::SemanticsRegistry;
         use crate::server::AppState;
         use datafusion::prelude::SessionContext;
         use skardi::engine::datafusion::DataFusionEngine;
         use std::path::PathBuf;
-        use std::sync::{Arc, RwLock};
+        use std::sync::Arc;
 
         let config = ServerConfig {
             pipelines: Default::default(),
@@ -364,14 +386,7 @@ mod tests {
         };
         let session_ctx = Arc::new(SessionContext::new());
         let engine = Arc::new(DataFusionEngine::new_with_arc(session_ctx.clone()));
-        AppState {
-            config: Arc::new(RwLock::new(config)),
-            engine,
-            session_ctx,
-            metrics: PipelineMetrics::new(),
-            auth_layer: AuthLayer::None,
-            jobs: None,
-        }
+        AppState::new(config, engine, session_ctx, AuthLayer::None, None)
     }
 
     #[tokio::test]
@@ -384,13 +399,12 @@ mod tests {
     async fn make_better_auth_state() -> AppState {
         use crate::auth::layer::AuthLayer;
         use crate::config::{CliArgs, ServerConfig};
-        use crate::metrics::PipelineMetrics;
         use crate::semantics::SemanticsRegistry;
         use crate::server::AppState;
         use datafusion::prelude::SessionContext;
         use skardi::engine::datafusion::DataFusionEngine;
         use std::path::PathBuf;
-        use std::sync::{Arc, RwLock};
+        use std::sync::Arc;
 
         unsafe {
             std::env::set_var("AUTH_SECRET", "test-secret-that-is-at-least-32-characters!");
@@ -418,14 +432,7 @@ mod tests {
         };
         let session_ctx = Arc::new(SessionContext::new());
         let engine = Arc::new(DataFusionEngine::new_with_arc(session_ctx.clone()));
-        AppState {
-            config: Arc::new(RwLock::new(config)),
-            engine,
-            session_ctx,
-            metrics: PipelineMetrics::new(),
-            auth_layer: layer,
-            jobs: None,
-        }
+        AppState::new(config, engine, session_ctx, layer, None)
     }
 
     #[tokio::test]

diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs
@@ -18,6 +18,7 @@ use skardi::sources::providers::redis::datasource::register_redis_tables;
 use skardi::sources::providers::seekdb::register_seekdb_tables;
 use skardi::sources::providers::sqlite::register_sqlite_tables;
 use skardi::sources::providers::sqlx::postgres::register_postgres_tables;
+use skardi::sources::sql_validator::{AdhocSqlPolicy, SqlValidatorConfig, validate_sql};
 use std::collections::HashMap;
 use std::path::Path;
 use std::path::PathBuf;
@@ -317,13 +318,13 @@ fn resolve_pipeline_files(path: Option<&PathBuf>) -> Result<Vec<PathBuf>> {
             let entry = entry.with_context(|| "Failed to read directory entry")?;
             let file_path = entry.path();
 
-            if file_path.is_file() {
-                if let Some(ext) = file_path.extension() {
-                    let ext = ext.to_string_lossy().to_lowercase();
-                    if ext == "yaml" || ext == "yml" {
-                        tracing::debug!("Found pipeline file: {:?}", file_path);
-                        pipeline_files.push(file_path);
-                    }
+            if file_path.is_file()
+                && let Some(ext) = file_path.extension()
+            {
+                let ext = ext.to_string_lossy().to_lowercase();
+                if ext == "yaml" || ext == "yml" {
+                    tracing::debug!("Found pipeline file: {:?}", file_path);
+                    pipeline_files.push(file_path);
                 }
             }
         }
@@ -524,12 +525,15 @@ pub async fn load_server_config(args: CliArgs) -> Result<ServerConfig> {
         }
     }
 
-    if args.jobs_path.is_some() && !job_files.is_empty() && jobs.is_empty() {
+    if let Some(jobs_path) = args.jobs_path.as_ref()
+        && !job_files.is_empty()
+        && jobs.is_empty()
+    {
         tracing::warn!(
             "--jobs {:?} scanned {} YAML file(s) but none had `kind: job` at the root; \
              /jobs/* endpoints will return 503 until at least one job definition loads. \
              Did you forget the `kind: job` discriminator?",
-            args.jobs_path.as_ref().expect("just checked is_some above"),
+            jobs_path,
             job_files.len(),
         );
     }
@@ -785,7 +789,7 @@ fn validate_data_sources(data_sources: &[DataSource]) -> Result<()> {
         {
             return Err(ConfigError::UnsupportedWriteMode {
                 name: source.name.clone(),
-                source_type: source.source_type.clone(),
+                source_type: source.source_type,
             }
             .into());
         }
@@ -861,19 +865,18 @@ fn validate_data_sources(data_sources: &[DataSource]) -> Result<()> {
                 }
             }
 
-            if source.source_type == DataSourceType::Dynamodb {
-                if let Some(value) = source
+            if source.source_type == DataSourceType::Dynamodb
+                && let Some(value) = source
                     .options
                     .as_ref()
                     .and_then(|o| o.get("allowed_tables"))
-                {
-                    let has_entry = value.split(',').any(|s| !s.trim().is_empty());
-                    if !has_entry {
-                        return Err(ConfigError::EmptyAllowedTables {
-                            name: source.name.clone(),
-                        }
-                        .into());
+            {
+                let has_entry = value.split(',').any(|s| !s.trim().is_empty());
+                if !has_entry {
+                    return Err(ConfigError::EmptyAllowedTables {
+                        name: source.name.clone(),
                     }
+                    .into());
                 }
             }
         }
@@ -944,24 +947,38 @@ fn validate_schema_types(_schema: &HashMap<String, String>) -> Result<()> {
     Ok(())
 }
 
+/// Build the access-mode map for every data source. Shared by both the
+/// trusted pipeline-load path and the untrusted `/query` policy below.
+pub fn validator_config_from_sources(data_sources: &[DataSource]) -> SqlValidatorConfig {
+    let mut validator_config = SqlValidatorConfig::new();
+    for ds in data_sources {
+        validator_config = validator_config.with_table(&ds.name, ds.access_mode);
+    }
+    validator_config
+}
+
+/// Build the statement policy for the untrusted ad-hoc `/query` endpoint:
+/// the access-mode map plus the reserved [`AUTH_SCHEMA`] denial (auth.users /
+/// auth.sessions register on the same `SessionContext` and hold live bearer
+/// tokens, so ad-hoc SQL must never reach them). The denial is scoped to this
+/// policy, so operator-authored pipeline SQL may still read auth tables.
+///
+/// Callers snapshot this once at startup into `AppState`. That is correct only
+/// as long as nothing mutates a source's `access_mode` at runtime — there is
+/// no such writer today. If one is ever added, it must rebuild this policy (or
+/// the snapshot will serve a stale, potentially more-permissive gate).
+pub fn adhoc_policy_from_sources(data_sources: &[DataSource]) -> AdhocSqlPolicy {
+    AdhocSqlPolicy::new(validator_config_from_sources(data_sources))
+        .with_denied_schema(crate::auth::bridge::AUTH_SCHEMA)
+}
+
 /// Validate pipeline SQL against data source access modes
 fn validate_pipeline_sql(
     pipeline_name: &str,
     sql: &str,
     data_sources: &[DataSource],
 ) -> Result<()> {
-    use skardi::sources::sql_validator::{SqlValidatorConfig, validate_sql};
-
-    // Build validator config from data sources
-    let mut validator_config = SqlValidatorConfig::new();
-    for ds in data_sources {
-        let mode = if ds.access_mode.is_read_write() {
-            skardi::sources::sql_validator::AccessMode::ReadWrite
-        } else {
-            skardi::sources::sql_validator::AccessMode::ReadOnly
-        };
-        validator_config = validator_config.with_table(&ds.name, mode);
-    }
+    let validator_config = validator_config_from_sources(data_sources);
 
     // Validate the SQL against access mode restrictions
     validate_sql(sql, &validator_config).map_err(|e| {
@@ -1092,15 +1109,15 @@ async fn register_data_source(
                     csv_read_options =
                         csv_read_options.has_header(has_header.parse::<bool>().unwrap_or(true));
                 }
-                if let Some(delimiter) = options.get("delimiter") {
-                    if let Some(delimiter_char) = delimiter.chars().next() {
-                        csv_read_options = csv_read_options.delimiter(delimiter_char as u8);
-                    }
+                if let Some(delimiter) = options.get("delimiter")
+                    && let Some(delimiter_char) = delimiter.chars().next()
+                {
+                    csv_read_options = csv_read_options.delimiter(delimiter_char as u8);
                 }
-                if let Some(schema_infer_max) = options.get("schema_infer_max_records") {
-                    if let Ok(max_records) = schema_infer_max.parse::<usize>() {
-                        csv_read_options = csv_read_options.schema_infer_max_records(max_records);
-                    }
+                if let Some(schema_infer_max) = options.get("schema_infer_max_records")
+                    && let Ok(max_records) = schema_infer_max.parse::<usize>()
+                {
+                    csv_read_options = csv_read_options.schema_infer_max_records(max_records);
                 }
             }
 
@@ -2238,7 +2255,7 @@ spec:
         use clap::Parser;
 
         // Test with single pipeline file and context
-        let args = CliArgs::try_parse_from(&[
+        let args = CliArgs::try_parse_from([
             "skardi-server",
             "--pipeline",
             "/path/to/pipeline.yaml",
@@ -2257,7 +2274,7 @@ spec:
         assert_eq!(args.port, 9000);
 
         // Test with pipeline directory
-        let args = CliArgs::try_parse_from(&["skardi-server", "--pipeline", "/path/to/pipelines/"])
+        let args = CliArgs::try_parse_from(["skardi-server", "--pipeline", "/path/to/pipelines/"])
             .unwrap();
 
         assert_eq!(
@@ -2268,7 +2285,7 @@ spec:
         assert_eq!(args.port, 8080); // default value
 
         // Test with no pipelines
-        let args = CliArgs::try_parse_from(&["skardi-server"]).unwrap();
+        let args = CliArgs::try_parse_from(["skardi-server"]).unwrap();
 
         assert!(args.pipeline_path.is_none());
         assert_eq!(args.ctx_file, None);