refactor(validator): type-split ad-hoc policy, share DataFusion's parser, tidy validator

Second-round review follow-ups (#3–#10):

- #4 Encode the trust boundary in the type. Ad-hoc /query now validates
  against a dedicated AdhocSqlPolicy (access modes + denied schemas);
  the trusted pipeline path keeps a bare SqlValidatorConfig. Reaching for
  validate_sql on untrusted input can no longer silently skip the
  allowlist + schema denial.
- #5 Add coverage proving the schema denial descends into indirect
  relations (set ops, scalar/IN/EXISTS subqueries) via visit_relations;
  document the residual (operator-defined views/federated aliases, which
  ad-hoc SQL cannot create) and its structural fix. See follow-up issue.
- #6 Parse via datafusion::sql::sqlparser (DataFusion enables the visitor
  feature); drop the standalone sqlparser dep entirely so validator and
  engine share one parser by construction — a DF bump is now a compile
  break, not silent parse divergence.
- #8 Single AUTH_SCHEMA const used at both the register and deny sites.
- #9 check_denied_schemas reports the table via extract_table_name
  (quote-stripped), matching WriteNotAllowed.
- #10 statement_keyword maps the AST variant to a &'static str instead of
  Display-rendering the whole statement on each rejection.
- #7 AppState::new derives the policy once; removes the copy-pasted
  Arc::new(validator_config_from_sources(..)) across ~9 sites.
- #3 Document the startup-snapshot invariant (no runtime access_mode
  writer) on the field and the builder.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: BtXin

Cargo.lock

@@ -64,7 +64,6 @@ rstest = "0.25.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_yaml = "0.9"
-sqlparser = { version = "0.59", features = ["visitor"] }
 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"] }

Cargo.toml

@@ -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");

crates/server/src/auth/bridge.rs

@@ -363,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(),
@@ -387,15 +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,
-            validator_config: Arc::new(crate::config::validator_config_from_sources(&[])),
-        }
+        AppState::new(config, engine, session_ctx, AuthLayer::None, None)
     }
 
     #[tokio::test]
@@ -408,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!");
@@ -442,15 +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,
-            validator_config: Arc::new(crate::config::validator_config_from_sources(&[])),
-        }
+        AppState::new(config, engine, session_ctx, layer, None)
     }
 
     #[tokio::test]

crates/server/src/auth/routes.rs

@@ -18,7 +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::{SqlValidatorConfig, validate_sql};
+use skardi::sources::sql_validator::{AdhocSqlPolicy, SqlValidatorConfig, validate_sql};
 use std::collections::HashMap;
 use std::path::Path;
 use std::path::PathBuf;
@@ -947,21 +947,31 @@ fn validate_schema_types(_schema: &HashMap<String, String>) -> Result<()> {
     Ok(())
 }
 
-/// Build a SQL validator config mapping every data source name to its
-/// configured access mode. Used at config load (pipeline SQL) and at
-/// request time (`POST /query`).
+/// 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 {
-    // The `auth` schema (auth.users / auth.sessions, registered on the same
-    // SessionContext) holds live bearer tokens; ad-hoc SQL must never reach
-    // it. The denial only applies to `validate_single_sql`, so pipeline SQL
-    // may still read auth tables.
-    let mut validator_config = SqlValidatorConfig::new().with_denied_schema("auth");
+    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,

crates/server/src/config.rs

@@ -778,7 +778,6 @@ pub async fn execute_pipeline_by_name(
 mod tests {
     use super::*;
     use crate::config::{CliArgs, DataSource, DataSourceType, ServerConfig};
-    use crate::metrics::PipelineMetrics;
     use crate::server::AppState;
     use arrow::array::{Int64Array, StringArray};
     use arrow::datatypes::{DataType, Field, Schema};
@@ -789,7 +788,7 @@ mod tests {
     use skardi::sources::AccessMode;
     use std::fs;
     use std::path::PathBuf;
-    use std::sync::{Arc, RwLock};
+    use std::sync::Arc;
     use tempfile::TempDir;
 
     async fn create_test_pipeline_with_params() -> StandardPipeline {
@@ -847,18 +846,13 @@ spec:
         let session_ctx = Arc::new(SessionContext::new());
         let engine = Arc::new(DataFusionEngine::new_with_arc(session_ctx.clone()));
 
-        let validator_config = Arc::new(crate::config::validator_config_from_sources(
-            &config.data_sources,
-        ));
-        AppState {
-            config: Arc::new(RwLock::new(config)),
+        AppState::new(
+            config,
             engine,
             session_ctx,
-            metrics: PipelineMetrics::new(),
-            auth_layer: crate::auth::layer::AuthLayer::None,
-            jobs: None,
-            validator_config,
-        }
+            crate::auth::layer::AuthLayer::None,
+            None,
+        )
     }
 
     fn create_test_record_batch() -> RecordBatch {
@@ -941,18 +935,13 @@ spec:
         let session_ctx_arc = Arc::new(session_ctx);
         let engine = Arc::new(DataFusionEngine::new_with_arc(session_ctx_arc.clone()));
 
-        let validator_config = Arc::new(crate::config::validator_config_from_sources(
-            &config.data_sources,
-        ));
-        let app_state = AppState {
-            config: Arc::new(RwLock::new(config)),
+        let app_state = AppState::new(
+            config,
             engine,
-            session_ctx: session_ctx_arc,
-            metrics: PipelineMetrics::new(),
-            auth_layer: crate::auth::layer::AuthLayer::None,
-            jobs: None,
-            validator_config,
-        };
+            session_ctx_arc,
+            crate::auth::layer::AuthLayer::None,
+            None,
+        );
 
         let request = ExecuteRequest {
             parameters: {

crates/server/src/pipeline_handlers.rs

@@ -68,7 +68,7 @@ pub async fn execute_query(
         None => DEFAULT_MAX_ROWS,
     };
 
-    let statement_kind = match validate_single_sql(&request.sql, &app_state.validator_config) {
+    let statement_kind = match validate_single_sql(&request.sql, &app_state.adhoc_policy) {
         Ok(kind) => kind,
         Err(e) => {
             tracing::info!("Rejected ad-hoc query: {}", e);

crates/server/src/query_handlers.rs

@@ -7,7 +7,7 @@ use datafusion::prelude::SessionContext;
 use skardi::engine::datafusion::DataFusionEngine;
 use skardi::jobs::{JobExecutor, JobStore, SqliteJobStore};
 use skardi::sources::DataSourceType;
-use skardi::sources::sql_validator::SqlValidatorConfig;
+use skardi::sources::sql_validator::AdhocSqlPolicy;
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::{Arc, RwLock};
@@ -55,10 +55,38 @@ pub struct AppState {
     /// Jobs executor + run ledger. `None` when the server was started
     /// without `--jobs`, which disables every `/jobs/*` endpoint.
     pub jobs: Option<Arc<JobExecutor>>,
-    /// Statement policy for the ad-hoc `/query` endpoint, built once at
-    /// startup from the configured data sources (there is no runtime
-    /// config writer, so a snapshot is sufficient).
-    pub validator_config: Arc<SqlValidatorConfig>,
+    /// Statement policy for the ad-hoc `/query` endpoint. Derived from the
+    /// data sources' access modes once at startup by [`AppState::new`]; see
+    /// [`crate::config::adhoc_policy_from_sources`] for why a snapshot is
+    /// safe (no runtime writer mutates access modes).
+    pub adhoc_policy: Arc<AdhocSqlPolicy>,
+}
+
+impl AppState {
+    /// Assemble the shared state, deriving the ad-hoc `/query` policy from the
+    /// config's data sources. Centralizing construction here keeps the derived
+    /// policy from drifting across the many call sites that build an
+    /// `AppState` (handlers, tests, integration harnesses).
+    pub fn new(
+        config: ServerConfig,
+        engine: Arc<DataFusionEngine>,
+        session_ctx: Arc<SessionContext>,
+        auth_layer: AuthLayer,
+        jobs: Option<Arc<JobExecutor>>,
+    ) -> Self {
+        let adhoc_policy = Arc::new(crate::config::adhoc_policy_from_sources(
+            &config.data_sources,
+        ));
+        Self {
+            config: Arc::new(RwLock::new(config)),
+            engine,
+            session_ctx,
+            metrics: PipelineMetrics::new(),
+            auth_layer,
+            jobs,
+            adhoc_policy,
+        }
+    }
 }
 
 /// Main server creation function - Primary public interface
@@ -221,20 +249,8 @@ pub async fn setup_app_state(config: ServerConfig) -> Result<AppState> {
         None
     };
 
-    let validator_config = Arc::new(crate::config::validator_config_from_sources(
-        &config.data_sources,
-    ));
-
     // Create shared application state with RwLock for runtime updates
-    let app_state = AppState {
-        config: Arc::new(RwLock::new(config)),
-        engine,
-        session_ctx: session_ctx_arc,
-        metrics: PipelineMetrics::new(),
-        auth_layer,
-        jobs: jobs_bundle,
-        validator_config,
-    };
+    let app_state = AppState::new(config, engine, session_ctx_arc, auth_layer, jobs_bundle);
 
     tracing::info!("Application state setup completed successfully");
 

crates/server/src/server.rs

@@ -15,14 +15,13 @@ use skardi::jobs::{JobDefinition, JobExecutor, JobStore, SqliteJobStore};
 use skardi::sources::DataSourceType;
 use std::collections::HashMap;
 use std::io::Write;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::time::Duration;
 use tempfile::TempDir;
 use tower::ServiceExt;
 
 use skardi_server::auth::layer::AuthLayer;
 use skardi_server::config::{CliArgs, ServerConfig};
-use skardi_server::metrics::PipelineMetrics;
 use skardi_server::semantics::SemanticsRegistry;
 use skardi_server::server::{AppState, configure_routes};
 
@@ -106,18 +105,7 @@ spec:
             port: 0,
         },
     };
-    let validator_config = Arc::new(skardi_server::config::validator_config_from_sources(
-        &config.data_sources,
-    ));
-    let state = AppState {
-        config: Arc::new(RwLock::new(config)),
-        engine,
-        session_ctx: Arc::clone(&ctx),
-        metrics: PipelineMetrics::new(),
-        auth_layer: AuthLayer::None,
-        jobs: executor,
-        validator_config,
-    };
+    let state = AppState::new(config, engine, Arc::clone(&ctx), AuthLayer::None, executor);
     (state, tmp)
 }
 

crates/server/tests/jobs_http.rs

@@ -15,13 +15,12 @@ use serde_json::{Value, json};
 use skardi::pipeline::pipeline::{Pipeline, StandardPipeline};
 use std::collections::HashMap;
 use std::io::Write;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use tempfile::TempDir;
 use tower::ServiceExt;
 
 use skardi_server::auth::layer::AuthLayer;
 use skardi_server::config::{CliArgs, ServerConfig};
-use skardi_server::metrics::PipelineMetrics;
 use skardi_server::semantics::SemanticsRegistry;
 use skardi_server::server::{AppState, configure_routes};
 
@@ -110,18 +109,7 @@ spec:
             port: 0,
         },
     };
-    let validator_config = Arc::new(skardi_server::config::validator_config_from_sources(
-        &config.data_sources,
-    ));
-    let state = AppState {
-        config: Arc::new(RwLock::new(config)),
-        engine,
-        session_ctx: Arc::clone(&ctx),
-        metrics: PipelineMetrics::new(),
-        auth_layer: AuthLayer::None,
-        jobs: None,
-        validator_config,
-    };
+    let state = AppState::new(config, engine, Arc::clone(&ctx), AuthLayer::None, None);
     (state, tmp)
 }