fix: restrict variables/resources/workspace-db mutations to non-operators

Also serialize quota checks against concurrent creates via advisory
transaction locks.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

backend/ee-repo-ref.txt

@@ -1 +1 @@
-e587df81d31638259efec13d0e68e54c890fdc2e
+c8ebea6ba778d233a1a25ce89a7f240b815fb04e

backend/windmill-api-workspaces/src/workspaces.rs

@@ -1693,6 +1693,11 @@ async fn create_pg_database(
     Path(w_id): Path<String>,
     Json(req): Json<CreatePgDatabaseRequest>,
 ) -> Result<String> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot create databases for security reasons".to_string(),
+        ));
+    }
     windmill_common::validate_dbname(&req.target_dbname)?;
 
     // Non-superadmin: restrict dbname to wm_fork_ prefix
@@ -1793,6 +1798,11 @@ async fn import_pg_database(
     Path(w_id): Path<String>,
     Json(req): Json<ImportPgDatabaseRequest>,
 ) -> Result<String> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot import databases for security reasons".to_string(),
+        ));
+    }
     if req.fork_behavior == DataTableForkBehavior::KeepOriginal {
         return Ok("No action needed for KeepOriginal behavior".to_string());
     }

backend/windmill-store/src/resources.rs

@@ -769,6 +769,11 @@ async fn create_resource(
     Query(q): Query<CreateResourceQuery>,
     Json(resource): Json<CreateResource>,
 ) -> Result<(StatusCode, String)> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot create resources for security reasons".to_string(),
+        ));
+    }
     check_scopes(&authed, || format!("resources:write:{}", resource.path))?;
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
         &w_id,
@@ -781,7 +786,22 @@ async fn create_resource(
     {
         return Err(Error::PermissionDenied(msg));
     }
+    let authed = maybe_refresh_folders(&resource.path, &w_id, authed, &db).await;
+
+    let mut tx = user_db.begin(&authed).await?;
+
     if *CLOUD_HOSTED {
+        // Serialize concurrent creates for this workspace so the quota check + insert
+        // behave atomically. Without the lock, bursts of requests can all observe a
+        // pre-limit count and collectively exceed the quota. The lock is on the user
+        // transaction (auto-released on commit) while the count goes through the
+        // admin pool so RLS doesn't hide workspace-wide rows from the quota check.
+        sqlx::query!(
+            "SELECT pg_advisory_xact_lock(hashtext('resource_quota:' || $1)::bigint)",
+            &w_id
+        )
+        .execute(&mut *tx)
+        .await?;
         let nb_resources = sqlx::query_scalar!(
             "SELECT COUNT(*) FROM resource WHERE workspace_id = $1",
             &w_id
@@ -795,9 +815,6 @@ async fn create_resource(
                 ));
         }
     }
-    let authed = maybe_refresh_folders(&resource.path, &w_id, authed, &db).await;
-
-    let mut tx = user_db.begin(&authed).await?;
 
     let update_if_exists = q.update_if_exists.unwrap_or(false);
     if !update_if_exists {
@@ -889,6 +906,11 @@ async fn delete_resource(
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> Result<String> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot delete resources for security reasons".to_string(),
+        ));
+    }
     let path = path.to_path();
 
     check_scopes(&authed, || format!("resources:write:{}", path))?;
@@ -1088,6 +1110,11 @@ async fn delete_resources_bulk(
     Path(w_id): Path<String>,
     Json(request): Json<BulkDeleteRequest>,
 ) -> JsonResult<Vec<String>> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot delete resources for security reasons".to_string(),
+        ));
+    }
     for path in &request.paths {
         check_scopes(&authed, || format!("resources:write:{}", path))?;
     }
@@ -1196,6 +1223,11 @@ async fn update_resource(
 ) -> Result<String> {
     use sql_builder::prelude::*;
 
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot update resources for security reasons".to_string(),
+        ));
+    }
     let path = path.to_path();
     check_scopes(&authed, || format!("resources:write:{}", path))?;
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
@@ -1388,6 +1420,11 @@ async fn update_resource_value(
     Path((w_id, path)): Path<(String, StripPath)>,
     Json(nv): Json<UpdateResource>,
 ) -> Result<String> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot update resources for security reasons".to_string(),
+        ));
+    }
     let path = path.to_path();
     check_scopes(&authed, || format!("resources:write:{}", path))?;
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
@@ -1578,6 +1615,11 @@ async fn create_resource_type(
     Path(w_id): Path<String>,
     Json(resource_type): Json<CreateResourceType>,
 ) -> Result<(StatusCode, String)> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot create resource types for security reasons".to_string(),
+        ));
+    }
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
         &w_id,
         AuditAuthorable::username(&authed),
@@ -1750,6 +1792,11 @@ async fn update_resource_type(
     Json(ns): Json<EditResourceType>,
 ) -> Result<String> {
     use sql_builder::prelude::*;
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot update resource types for security reasons".to_string(),
+        ));
+    }
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
         &w_id,
         AuditAuthorable::username(&authed),

backend/windmill-store/src/variables.rs

@@ -390,6 +390,11 @@ async fn create_variable(
     Query(AlreadyEncrypted { already_encrypted }): Query<AlreadyEncrypted>,
     Json(variable): Json<CreateVariable>,
 ) -> Result<(StatusCode, String)> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot create variables for security reasons".to_string(),
+        ));
+    }
     check_scopes(&authed, || format!("variables:write:{}", variable.path))?;
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
         &w_id,
@@ -403,7 +408,30 @@ async fn create_variable(
         return Err(Error::PermissionDenied(msg));
     }
 
+    let authed = maybe_refresh_folders(&variable.path, &w_id, authed, &db).await;
+
+    check_path_conflict(&db, &w_id, &variable.path).await?;
+    let value = if variable.is_secret && !already_encrypted.unwrap_or(false) {
+        // Use secret backend for encryption (supports both DB and Vault)
+        store_secret_value(&db, &w_id, &variable.path, &variable.value).await?
+    } else {
+        variable.value
+    };
+
+    let mut tx = user_db.begin(&authed).await?;
+
     if *CLOUD_HOSTED {
+        // Serialize concurrent creates for this workspace so the quota check + insert
+        // behave atomically. Without the lock, bursts of requests can all observe a
+        // pre-limit count and collectively exceed the quota. The lock is on the user
+        // transaction (auto-released on commit) while the count goes through the
+        // admin pool so RLS doesn't hide workspace-wide rows from the quota check.
+        sqlx::query!(
+            "SELECT pg_advisory_xact_lock(hashtext('variable_quota:' || $1)::bigint)",
+            &w_id
+        )
+        .execute(&mut *tx)
+        .await?;
         let nb_variables = sqlx::query_scalar!(
             "SELECT COUNT(*) FROM variable WHERE workspace_id = $1",
             &w_id
@@ -417,17 +445,6 @@ async fn create_variable(
                 ));
         }
     }
-    let authed = maybe_refresh_folders(&variable.path, &w_id, authed, &db).await;
-
-    check_path_conflict(&db, &w_id, &variable.path).await?;
-    let value = if variable.is_secret && !already_encrypted.unwrap_or(false) {
-        // Use secret backend for encryption (supports both DB and Vault)
-        store_secret_value(&db, &w_id, &variable.path, &variable.value).await?
-    } else {
-        variable.value
-    };
-
-    let mut tx = user_db.begin(&authed).await?;
 
     sqlx::query!(
         "INSERT INTO variable
@@ -500,6 +517,11 @@ async fn delete_variable(
     Extension(webhook): Extension<WebhookShared>,
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> Result<String> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot delete variables for security reasons".to_string(),
+        ));
+    }
     let path = path.to_path();
 
     check_scopes(&authed, || format!("variables:write:{}", path))?;
@@ -646,6 +668,11 @@ async fn delete_variables_bulk(
     Path(w_id): Path<String>,
     Json(request): Json<BulkDeleteRequest>,
 ) -> JsonResult<Vec<String>> {
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot delete variables for security reasons".to_string(),
+        ));
+    }
     for path in &request.paths {
         check_scopes(&authed, || format!("variables:write:{}", path))?;
     }
@@ -796,6 +823,12 @@ async fn update_variable(
 ) -> Result<String> {
     use sql_builder::prelude::*;
 
+    if authed.is_operator {
+        return Err(Error::NotAuthorized(
+            "Operators cannot update variables for security reasons".to_string(),
+        ));
+    }
+
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
         &w_id,
         AuditAuthorable::username(&authed),