fix: let operators use wmill.datatable() from within running jobs (#10931)

* fix: let operators use wmill.datatable() from within running jobs

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RHR4fytgt6m4q37WCXs2Rp

* fix: refuse content-driven redirects and deferral in the operator datatable exemption

* fix: check the datatable exemption against the expanded query, not the raw content

* fix: fail closed on a language-overriding expansion and state the exemption's real scope

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

backend/tests/fixtures/inline_preview_auth.sql

@@ -2,7 +2,9 @@
 -- Layered on top of `base` (which provides test-workspace and the non-operator
 -- `test-user-2`/SECRET_TOKEN_2). Adds an Operator member so we can assert that
 -- Operators cannot reach the arbitrary-code inline preview path
--- (`POST /jobs/run_inline/preview`).
+-- (`POST /jobs/run_inline/preview`) with their own token, plus two deployed script
+-- jobs of the operator: one running, so we can assert that its WM_TOKEN can, and
+-- one queued but not yet pulled, so we can assert that "queued" is not enough.
 
 INSERT INTO password(email, password_hash, login_type, super_admin, verified, name)
     VALUES ('operator@windmill.dev', 'not-a-real-hash', 'password', false, true, 'Operator User');
@@ -12,3 +14,11 @@ INSERT INTO usr(workspace_id, email, username, is_admin, operator, role) VALUES
 
 INSERT INTO token(token_hash, token_prefix, token, email, label, super_admin) VALUES
 	(encode(sha256('OPERATOR_TOKEN'::bytea), 'hex'), 'OPERATOR_T', 'OPERATOR_TOKEN', 'operator@windmill.dev', 'operator token', false);
+
+INSERT INTO v2_job(id, workspace_id, kind, runnable_path, created_by, permissioned_as, permissioned_as_email) VALUES
+	('2aa0c0de-0000-4000-8000-000000000001', 'test-workspace', 'script', 'u/test-user/deployed', 'operator-user', 'u/operator-user', 'operator@windmill.dev'),
+	('2aa0c0de-0000-4000-8000-000000000002', 'test-workspace', 'script', 'u/test-user/deployed', 'operator-user', 'u/operator-user', 'operator@windmill.dev');
+
+INSERT INTO v2_job_queue(id, workspace_id, scheduled_for, running) VALUES
+	('2aa0c0de-0000-4000-8000-000000000001', 'test-workspace', now(), true),
+	('2aa0c0de-0000-4000-8000-000000000002', 'test-workspace', now(), false);

backend/tests/inline_preview_auth.rs

@@ -9,16 +9,33 @@
 //! was the incomplete-fix residual of CVE-2026-22683, whose v1.615.0 patch only
 //! covered the entity-CRUD endpoints and left this direct inline-exec sink open.
 //!
+//! The guard on both routes has one exemption: `wmill.datatable()` called from
+//! inside a job the operator is running. Operators can only run deployed code,
+//! so a request the job's WM_TOKEN authenticates comes from code a non-operator
+//! authored, and the exemption is limited to the request shape the helper sends
+//! (PostgreSQL against a `datatable://` database) so a leaked WM_TOKEN cannot
+//! be replayed to run anything else.
+//!
 //! This test pins down:
-//!   - an Operator is rejected by the operator guard (the core fix; pre-fix this
-//!     reached the inline executor instead of returning 401), and
+//!   - an Operator's own token is rejected by the operator guard (the core fix;
+//!     pre-fix this reached the inline executor instead of returning 401),
 //!   - a regular non-operator passes the guard (the fix must not over-block the
 //!     legitimate inline preview flow): in the test harness the worker inline
 //!     utils are not registered, so a caller past the guard gets the distinct
-//!     "worker inline functions" error rather than the operator rejection.
+//!     "worker inline functions" error rather than the operator rejection,
+//!   - an Operator's job token passes the guard for a datatable query while its
+//!     job is running, on the inline route and on the `/jobs/run/preview`
+//!     fallback the SDKs use when the worker has no internal server,
+//!   - the same token is rejected for any other payload (in-process DuckDB, or a
+//!     `-- database` directive redirecting the query, whether written literally or
+//!     reached through a `WM_INTERNAL_DB` marker) and for a deferred run,
+//!   - an Operator's job token for a job that is not running, whether finished or
+//!     merely queued, is rejected.
 
 use serde_json::json;
 use sqlx::{Pool, Postgres};
+use windmill_common::auth::create_jwt_token;
+use windmill_common::db::Authed;
 use windmill_test_utils::*;
 
 fn client() -> reqwest::Client {
@@ -38,11 +55,65 @@ fn inline_preview_body() -> serde_json::Value {
     })
 }
 
+/// The request `wmill.datatable("main")` sends: PostgreSQL against `datatable://main`.
+fn datatable_query_body() -> serde_json::Value {
+    json!({
+        "language": "postgresql",
+        "content": "SELECT 1 AS x;",
+        "args": { "database": "datatable://main" }
+    })
+}
+
+/// Mint the WM_TOKEN a job hands its own code: an internally-signed job JWT
+/// (note the `job_id` claim) for the fixture's operator, exactly as the worker
+/// issues it when the operator runs a deployed script.
+async fn operator_job_token(job_id: uuid::Uuid) -> String {
+    let authed = Authed {
+        email: "operator@windmill.dev".to_string(),
+        username: "operator-user".to_string(),
+        is_admin: false,
+        is_operator: true,
+        groups: vec![],
+        folders: vec![],
+        scopes: None,
+        token_prefix: None,
+    };
+    create_jwt_token(
+        authed,
+        "test-workspace",
+        3600,
+        Some(job_id),
+        Some("ephemeral-script".to_string()),
+        None,
+        None,
+    )
+    .await
+    .expect("mint operator job token")
+}
+
 const OPERATOR_GUARD_MSG: &str = "Operators cannot run preview jobs";
 
+/// The fixture's deployed-script jobs of the operator: one running, one queued.
+const RUNNING_JOB_ID: &str = "2aa0c0de-0000-4000-8000-000000000001";
+const QUEUED_JOB_ID: &str = "2aa0c0de-0000-4000-8000-000000000002";
+
+async fn post(url: &str, token: &str, body: &serde_json::Value) -> (u16, String) {
+    let resp = authed(client().post(url), token)
+        .json(body)
+        .send()
+        .await
+        .expect("request");
+    let status = resp.status().as_u16();
+    let body = resp.text().await.expect("body");
+    (status, body)
+}
+
 #[sqlx::test(fixtures("base", "inline_preview_auth"))]
 async fn test_inline_preview_authorization(db: Pool<Postgres>) -> anyhow::Result<()> {
     initialize_tracing().await;
+    // The server decodes WM_TOKENs with the same in-process JWT secret, so
+    // setting it once lets us mint valid ones below.
+    set_jwt_secret().await;
 
     let server = ApiServer::start(db.clone()).await?;
     let port = server.addr.port();
@@ -51,12 +122,7 @@ async fn test_inline_preview_authorization(db: Pool<Postgres>) -> anyhow::Result
     // 1. CORE REGRESSION: an Operator must be rejected by the operator guard.
     //    Pre-fix this fell through to the inline executor (arbitrary code
     //    execution); post-fix it returns 401 with the operator guard message.
-    let resp = authed(client().post(&url), "OPERATOR_TOKEN")
-        .json(&inline_preview_body())
-        .send()
-        .await?;
-    let status = resp.status();
-    let body = resp.text().await?;
+    let (status, body) = post(&url, "OPERATOR_TOKEN", &inline_preview_body()).await;
     assert_eq!(
         status, 401,
         "Operator must be rejected from inline preview (got {status}): {body}"
@@ -71,12 +137,7 @@ async fn test_inline_preview_authorization(db: Pool<Postgres>) -> anyhow::Result
     //    the worker inline utils, so the request proceeds past the guard and
     //    fails later with the distinct "worker inline functions" error — proving
     //    the operator guard did not reject it.
-    let resp = authed(client().post(&url), "SECRET_TOKEN_2")
-        .json(&inline_preview_body())
-        .send()
-        .await?;
-    let status = resp.status();
-    let body = resp.text().await?;
+    let (status, body) = post(&url, "SECRET_TOKEN_2", &inline_preview_body()).await;
     assert_ne!(
         status, 401,
         "non-operator must not be blocked by the operator guard (got {status}): {body}"
@@ -86,5 +147,113 @@ async fn test_inline_preview_authorization(db: Pool<Postgres>) -> anyhow::Result
         "non-operator must not hit the operator guard, got: {body}"
     );
 
+    // 3. The WM_TOKEN of a deployed-script job the Operator is running passes the
+    //    guard for a datatable query: this is `wmill.datatable()` called from
+    //    inside that job. As in 2, the harness then fails with the "worker inline
+    //    functions" error.
+    let running_job_token =
+        operator_job_token(uuid::Uuid::parse_str(RUNNING_JOB_ID).unwrap()).await;
+    let (status, body) = post(&url, &running_job_token, &datatable_query_body()).await;
+    assert_ne!(
+        status, 401,
+        "operator job token of a running job must pass the guard for a datatable query (got {status}): {body}"
+    );
+    assert!(
+        !body.contains(OPERATOR_GUARD_MSG),
+        "operator job token of a running job must not hit the operator guard, got: {body}"
+    );
+
+    // 4. The same token is rejected for any other payload: the exemption covers
+    //    the datatable request shape only, never in-process DuckDB, and never a
+    //    `-- database` directive, which the executor honors over `args.database`.
+    let mut redirected = datatable_query_body();
+    redirected["content"] = json!("-- database u/test-user/other_db\nSELECT 1 AS x;");
+    let mut to_s3 = datatable_query_body();
+    to_s3["content"] = json!("-- s3\nSELECT 1 AS x;");
+    let mut resource_db = datatable_query_body();
+    resource_db["args"]["database"] = json!("$res:u/test-user/other_db");
+    // A marker is a single line the directive regexes cannot match; the directive only
+    // appears once the executor expands it, so the guard must check the expansion.
+    let mut marker = datatable_query_body();
+    marker["content"] = json!(concat!(
+        r#"-- WM_INTERNAL_DB_SELECT {"table":"t","columnDefs":[{"field":"id","datatype":"int4"}],"#,
+        r#""whereClause":"true\n-- database u/test-user/other_db\n AND true"}"#
+    ));
+    for (label, payload) in [
+        ("DuckDB", inline_preview_body()),
+        ("database directive", redirected),
+        ("s3 directive", to_s3),
+        ("resource database", resource_db),
+        ("marker-expanded database directive", marker),
+    ] {
+        let (status, body) = post(&url, &running_job_token, &payload).await;
+        assert_eq!(
+            status, 401,
+            "operator job token must be rejected for a {label} payload (got {status}): {body}"
+        );
+        assert!(
+            body.contains(OPERATOR_GUARD_MSG),
+            "rejection for a {label} payload must be the operator guard, got: {body}"
+        );
+    }
+
+    // 5. An Operator's job token whose job is not running is rejected like the
+    //    operator's own token, whether the job is over (no queue row) or merely
+    //    queued: a WM_TOKEN that leaked through logs cannot be replayed once the
+    //    job is over.
+    for (label, job_id) in [
+        ("finished", uuid::Uuid::new_v4()),
+        ("queued", uuid::Uuid::parse_str(QUEUED_JOB_ID).unwrap()),
+    ] {
+        let token = operator_job_token(job_id).await;
+        let (status, body) = post(&url, &token, &datatable_query_body()).await;
+        assert_eq!(
+            status, 401,
+            "operator job token of a {label} job must be rejected (got {status}): {body}"
+        );
+        assert!(
+            body.contains(OPERATOR_GUARD_MSG),
+            "rejection for a {label} job must be the operator guard, got: {body}"
+        );
+    }
+
+    // 6. The SDKs fall back to `/jobs/run/preview` when the worker has no internal
+    //    server (agent workers). The same exemption applies there: the running
+    //    job's token queues the datatable query (201 with the job id), the
+    //    operator's own token is still refused.
+    let fallback_url = format!("http://localhost:{port}/api/w/test-workspace/jobs/run/preview");
+    let (status, body) = post(&fallback_url, &running_job_token, &datatable_query_body()).await;
+    assert_eq!(
+        status, 201,
+        "operator job token of a running job must queue a datatable preview (got {status}): {body}"
+    );
+    let (status, body) = post(&fallback_url, "OPERATOR_TOKEN", &datatable_query_body()).await;
+    assert_eq!(
+        status, 401,
+        "Operator must be rejected from the preview fallback (got {status}): {body}"
+    );
+    assert!(
+        body.contains(OPERATOR_GUARD_MSG),
+        "rejection must be the operator guard, got: {body}"
+    );
+
+    // 7. A deferred run on the fallback would outlive the running job the
+    //    exemption keys off, so the running job's token cannot schedule one.
+    for deferral in [
+        "scheduled_in_secs=86400",
+        "scheduled_for=2099-01-01T00:00:00Z",
+    ] {
+        let deferred_url = format!("{fallback_url}?{deferral}");
+        let (status, body) = post(&deferred_url, &running_job_token, &datatable_query_body()).await;
+        assert_eq!(
+            status, 401,
+            "operator job token must not schedule a deferred preview with {deferral} (got {status}): {body}"
+        );
+        assert!(
+            body.contains(OPERATOR_GUARD_MSG),
+            "rejection for {deferral} must be the operator guard, got: {body}"
+        );
+    }
+
     Ok(())
 }

backend/windmill-api/src/jobs.rs

@@ -18,6 +18,7 @@ use quick_cache::sync::Cache;
 use serde_json::value::RawValue;
 use serde_json::Value;
 use sha2::{Digest, Sha256};
+use std::borrow::Cow;
 use std::collections::HashMap;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -108,6 +109,7 @@ use windmill_common::{
     flows::{add_virtual_items_if_necessary, resolve_maybe_value, FlowValue},
     jobs::{script_path_to_payload, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode},
     oauth2::HmacSha256,
+    query_builders,
     scripts::{ScriptHash, ScriptLang},
     users::username_to_permissioned_as,
     utils::{not_found_if_none, now_from_db, paginate, require_admin, Pagination, StripPath},
@@ -8131,6 +8133,79 @@ pub async fn run_wait_result_flow_by_version(
     .await
 }
 
+/// Whether request-supplied SQL from an operator may run. Operators can only run deployed
+/// code, so a request their job token (`WM_TOKEN`) authenticates comes from code a
+/// non-operator authored. The job must still be running, and the request must have the
+/// shape `wmill.datatable()` sends (PostgreSQL against a `datatable://` database), so a
+/// WM_TOKEN that leaked into job logs cannot be replayed to reach another target while the
+/// job lives, in particular DuckDB, which runs in-process in the worker.
+///
+/// What it does permit is any statement against the workspace's data tables, writes and DDL
+/// included: the helper's body is an unrestricted SQL template and data tables carry no
+/// per-user ACL. Narrowing that is a separate decision from this exemption.
+///
+/// The database argument is only half the target: the executor honors a `-- database`
+/// directive in the SQL over it, and `-- s3` redirects the result set, so both are refused.
+/// Check them against the code the executor runs rather than the request's `content`, which
+/// is not the same string once a `WM_INTERNAL_DB` marker expands.
+async fn operator_may_run_datatable_query(
+    db: &DB,
+    w_id: &str,
+    job_id: Option<Uuid>,
+    language: Option<&ScriptLang>,
+    content: &str,
+    args: Option<&HashMap<String, Box<JsonRawValue>>>,
+) -> error::Result<bool> {
+    let Some(job_id) = job_id else {
+        return Ok(false);
+    };
+    if language != Some(&ScriptLang::Postgresql) {
+        return Ok(false);
+    }
+    // Parse the directives out of the code the executor actually runs: it expands a
+    // `WM_INTERNAL_DB` marker first, and a directive can be embedded in the expansion.
+    // An expansion that overrides the language would run something other than the SQL the
+    // language check above cleared, so it is refused along with a malformed marker.
+    let executed =
+        match query_builders::try_expand_internal_db_query(content, &ScriptLang::Postgresql) {
+            Some(Ok(expanded)) if expanded.language_override.is_none() => Cow::Owned(expanded.code),
+            Some(_) => return Ok(false),
+            None => Cow::Borrowed(content),
+        };
+    if windmill_parser_sql::parse_db_resource(&executed).is_some()
+        || !matches!(windmill_parser_sql::parse_s3_mode(&executed), Ok(None))
+    {
+        return Ok(false);
+    }
+    let targets_datatable = args
+        .and_then(|args| args.get("database"))
+        .and_then(|database| serde_json::from_str::<String>(database.get()).ok())
+        .is_some_and(|database| database.starts_with("datatable://"));
+    if !targets_datatable {
+        return Ok(false);
+    }
+    Ok(sqlx::query_scalar!(
+        "SELECT running AS \"running!\" FROM v2_job_queue WHERE id = $1 AND workspace_id = $2",
+        job_id,
+        w_id
+    )
+    .fetch_optional(db)
+    .await?
+    .unwrap_or(false))
+}
+
+/// The refusal an operator gets from a preview route. Inside a job the caller never ran a
+/// preview themselves, so name the one thing the job's token may do.
+fn operator_preview_refusal(job_id: Option<Uuid>) -> error::Error {
+    let reason = if job_id.is_some() {
+        "Operators cannot run preview jobs for security reasons: from inside a job, an \
+         operator may only run a wmill.datatable() query while that job is running"
+    } else {
+        "Operators cannot run preview jobs for security reasons"
+    };
+    error::Error::NotAuthorized(reason.to_string())
+}
+
 async fn run_preview_script(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
@@ -8142,9 +8217,20 @@ async fn run_preview_script(
     #[cfg(feature = "enterprise")]
     check_license_key_valid().await?;
     if authed.is_operator {
-        return Err(error::Error::NotAuthorized(
-            "Operators cannot run preview jobs for security reasons".to_string(),
-        ));
+        // A deferred run would outlive the running job the exemption keys off.
+        if run_query.get_scheduled_for(&db).await?.is_some()
+            || !operator_may_run_datatable_query(
+                &db,
+                &w_id,
+                authed.job_id,
+                preview.language.as_ref(),
+                preview.content.as_deref().unwrap_or_default(),
+                preview.args.as_ref(),
+            )
+            .await?
+        {
+            return Err(operator_preview_refusal(authed.job_id));
+        }
     }
     // Preview runs arbitrary, request-supplied code. require_path_read_access_for_preview
     // only checks folder/namespace *read* access (and is a no-op when path is null), so a
@@ -8239,13 +8325,21 @@ async fn run_inline_preview_script(
     Path(w_id): Path<String>,
     Json(preview): Json<PreviewInline>,
 ) -> error::Result<Response> {
-    // Same arbitrary-code class as run_preview_script: operators are blocked from
-    // running request-supplied code, and a narrowly-scoped token must not escape
-    // its scope through inline preview.
-    if authed.is_operator {
-        return Err(error::Error::NotAuthorized(
-            "Operators cannot run preview jobs for security reasons".to_string(),
-        ));
+    // Same arbitrary-code class as run_preview_script, and every worker and standalone
+    // server exposes this route, so an operator is refused on the same terms. A
+    // narrowly-scoped token must not escape its scope through inline preview either.
+    if authed.is_operator
+        && !operator_may_run_datatable_query(
+            &db,
+            &w_id,
+            job_id,
+            Some(&preview.language),
+            &preview.content,
+            preview.args.as_ref(),
+        )
+        .await?
+    {
+        return Err(operator_preview_refusal(job_id));
     }
     check_scopes(&authed, || format!("jobs:run"))?;
     if let Some(job_id) = job_id {