kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: backend arg schema validation (#5455)

Browse Source 
* Make schema validation struct

Schema Validation rules that are constructed from the schema or from the
MainArgSig(TODO).

* Make other validator builder

* Fail dependency job like with lockfile failing for schema validator

* Add last types + tests

* Remove unused dependency

* fix typos

* Migration ID was colliding with another, changed it manually

* Add Oneof + other fixes

* fix: cache for querying scripts correclty handles ScriptMetadata

* Add cache for schema validation from main arg sig

* Prepare sqlx

* Remove default features

* Feature flags

* Fix down migration table name

* cleanup: put validation logic inside a function

* Refactor to cache the should_validate boolean

Changed the schemavalidators cache to take in an
Option<SchemaValidator>, effectively storing the `should_validate_schema` information.

Also pass the schema when avaialble to construct the schema validator

* Add other job kinds to u8 cache key just in case

* Only cache if not preview
This commit is contained in:
wendrul
2025-03-19 19:41:10 +01:00
committed by GitHub
parent 27dfbdc7d1
commit 299c1430eb
18 changed files with 1081 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
backend/.sqlx/query-d55eabd42559893bdf91c5fdab1ae58006013be26fcb9367b3c6830f38749799.json → backend/.sqlx/query-03ae5b1c912b13a8a7aadf50cb4984a2ea952e782fd52eb3088454690bd13dd1.json generated
View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT content AS \"content!: String\",\n lock AS \"lock: String\", language AS \"language: Option<ScriptLang>\", envs AS \"envs: Vec<String>\", codebase LIKE '%.tar' as use_tar FROM script WHERE hash = $1 LIMIT 1",
"query": "SELECT content AS \"content!: String\",\n lock AS \"lock: String\", language AS \"language: Option<ScriptLang>\", envs AS \"envs: Vec<String>\", schema AS \"schema: String\", schema_validation AS \"schema_validation: bool\", codebase LIKE '%.tar' as use_tar FROM script WHERE hash = $1 LIMIT 1",
"describe": {
"columns": [
{
@@ -52,6 +52,16 @@
},
{
"ordinal": 4,
"name": "schema: String",
"type_info": "Json"
},
{
"ordinal": 5,
"name": "schema_validation: bool",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "use_tar",
"type_info": "Bool"
}
@@ -66,8 +76,10 @@
true,
false,
true,
true,
false,
null
]
},
"hash": "d55eabd42559893bdf91c5fdab1ae58006013be26fcb9367b3c6830f38749799"
"hash": "03ae5b1c912b13a8a7aadf50cb4984a2ea952e782fd52eb3088454690bd13dd1"
}

30
backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json generated
View File

@@ -130,28 +130,28 @@
},
{
"ordinal": 25,
"name": "ai_models",
"type_info": "VarcharArray"
},
{
"ordinal": 26,
"name": "code_completion_model",
"type_info": "Varchar"
},
{
"ordinal": 27,
"name": "teams_command_script",
"type_info": "Text"
},
{
"ordinal": 28,
"ordinal": 26,
"name": "teams_team_id",
"type_info": "Text"
},
{
"ordinal": 29,
"ordinal": 27,
"name": "teams_team_name",
"type_info": "Text"
},
{
"ordinal": 28,
"name": "ai_models",
"type_info": "VarcharArray"
},
{
"ordinal": 29,
"name": "code_completion_model",
"type_info": "Varchar"
}
],
"parameters": {
@@ -185,10 +185,10 @@
true,
true,
true,
true,
true,
true,
false,
true,
true,
true,
true
]
},

30
backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json generated
View File

@@ -130,28 +130,28 @@
},
{
"ordinal": 25,
"name": "ai_models",
"type_info": "VarcharArray"
},
{
"ordinal": 26,
"name": "code_completion_model",
"type_info": "Varchar"
},
{
"ordinal": 27,
"name": "teams_command_script",
"type_info": "Text"
},
{
"ordinal": 28,
"ordinal": 26,
"name": "teams_team_id",
"type_info": "Text"
},
{
"ordinal": 29,
"ordinal": 27,
"name": "teams_team_name",
"type_info": "Text"
},
{
"ordinal": 28,
"name": "ai_models",
"type_info": "VarcharArray"
},
{
"ordinal": 29,
"name": "code_completion_model",
"type_info": "Varchar"
}
],
"parameters": {
@@ -185,10 +185,10 @@
true,
true,
true,
true,
true,
true,
false,
true,
true,
true,
true
]
},

7
backend/.sqlx/query-bc00efe12901bc7b49c7892d2d9da675e0eaf5e6cc013149740bc778987a1006.json → backend/.sqlx/query-d15f02f090b8d1a7e816fe11b2e0867540ab6bb02ac6bf82decc220dce0ab048.json generated
View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32)",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33)",
"describe": {
"columns": [],
"parameters": {
@@ -77,10 +77,11 @@
"Bool",
"Varchar",
"Bool",
"Text"
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "bc00efe12901bc7b49c7892d2d9da675e0eaf5e6cc013149740bc778987a1006"
"hash": "d15f02f090b8d1a7e816fe11b2e0867540ab6bb02ac6bf82decc220dce0ab048"
}

2
backend/Cargo.toml
View File

@@ -357,4 +357,4 @@ tokio-tungstenite = { version = "0.24.0", features = ["native-tls"] }
tree-sitter = {version = "0.23.0", features = []}
tree-sitter-c-sharp = "0.23.0"
oracle = { version = "0.6.3", features = ["chrono"] }
rumqttc = { version = "0.24.0", features = ["use-native-tls"]}
rumqttc = { version = "0.24.0", features = ["use-native-tls"]}

3
backend/migrations/20250205141517_backend_schema_validation.down.sql Normal file
View File

@@ -0,0 +1,3 @@
-- Add down migration script here
ALTER TABLE script
DROP COLUMN schema_validation;

3
backend/migrations/20250205141517_backend_schema_validation.up.sql Normal file
View File

@@ -0,0 +1,3 @@
-- Add up migration script here
ALTER TABLE script
ADD COLUMN schema_validation BOOLEAN NOT NULL DEFAULT FALSE;

2
backend/parsers/windmill-parser-py/src/lib.rs
View File

@@ -57,6 +57,8 @@ fn filter_non_main(code: &str, main_name: &str) -> String {
return filtered_code;
}
/// skip_params is a micro optimization for when we just want to find the main
/// function without parsing all the params.
pub fn parse_python_signature(
code: &str,
override_main: Option<String>,

2
backend/parsers/windmill-parser-ts/src/lib.rs
View File

@@ -131,6 +131,8 @@ pub fn parse_expr_for_ids(code: &str) -> anyhow::Result<Vec<(String, String)>> {
Ok(visitor.idents.into_iter().collect())
}
/// skip_params is a micro optimization for when we just want to find the main
/// function without parsing all the params.
pub fn parse_deno_signature(
code: &str,
skip_dflt: bool,

23
backend/windmill-api/src/scripts.rs
View File

@@ -44,20 +44,12 @@ use windmill_audit::ActionKind;
use windmill_common::error::to_anyhow;
use windmill_common::{
db::UserDB,
error::{Error, JsonResult, Result},
jobs::JobPayload,
schedule::Schedule,
scripts::{
db::UserDB, error::{Error, JsonResult, Result}, jobs::JobPayload, schedule::Schedule, schema::should_validate_schema, scripts::{
to_i64, HubScript, ListScriptQuery, ListableScript, NewScript, Schema, Script, ScriptHash,
ScriptHistory, ScriptHistoryUpdate, ScriptKind, ScriptLang, ScriptWithStarred,
},
users::username_to_permissioned_as,
utils::{
}, users::username_to_permissioned_as, utils::{
not_found_if_none, paginate, query_elems_from_hub, require_admin, Pagination, StripPath,
},
worker::to_raw_value,
HUB_BASE_URL,
}, worker::to_raw_value, HUB_BASE_URL
};
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
use windmill_parser_ts::remove_pinned_imports;
@@ -645,6 +637,8 @@ async fn create_script_internal<'c>(
ns.language.clone()
};
let validate_schema = should_validate_schema(&ns.content, &ns.language);
let (no_main_func, has_preprocessor) = match lang {
ScriptLang::Bun | ScriptLang::Bunnative | ScriptLang::Deno | ScriptLang::Nativets => {
let args = windmill_parser_ts::parse_deno_signature(&ns.content, true, true, None)?;
@@ -662,8 +656,8 @@ async fn create_script_internal<'c>(
content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \
dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \
delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32)",
delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33)",
&w_id,
&hash.0,
ns.path,
@@ -699,7 +693,8 @@ async fn create_script_internal<'c>(
Some(&authed.email)
} else {
None
}
},
validate_schema,
)
.execute(&mut *tx)
.await?;

58
backend/windmill-common/src/cache.rs
View File

@@ -7,9 +7,13 @@
//! This shall only be used for testing, e.g. [`sqlx::test`] spawn a database per test,
//! and there is only one test per thread, so using thread-local cache avoid unexpected results.
use anyhow::anyhow;
use crate::{
apps::AppScriptId, error, flows::FlowNodeId, flows::FlowValue, scripts::ScriptHash,
scripts::ScriptLang,
apps::AppScriptId,
error,
flows::{FlowNodeId, FlowValue},
schema::SchemaValidator,
scripts::{ScriptHash, ScriptLang},
};
#[cfg(feature = "scoped_cache")]
@@ -307,6 +311,8 @@ pub struct ScriptMetadata {
pub language: Option<ScriptLang>,
pub envs: Option<Vec<String>>,
pub codebase: Option<String>,
pub schema: Option<String>,
pub schema_validator: Option<SchemaValidator>,
}
#[derive(Debug)]
@@ -529,6 +535,8 @@ pub mod script {
lock AS \"lock: String\", \
language AS \"language: Option<ScriptLang>\", \
envs AS \"envs: Vec<String>\", \
schema AS \"schema: String\", \
schema_validation AS \"schema_validation: bool\", \
codebase LIKE '%.tar' as use_tar \
FROM script WHERE hash = $1 LIMIT 1",
hash.0
@@ -537,23 +545,36 @@ pub mod script {
.await
.map_err(Into::into)
.and_then(unwrap_or_error(&loc, "Script", hash))
.map(|r| RawScript {
content: r.content,
lock: r.lock,
meta: Some(ScriptMetadata {
language: r.language,
envs: r.envs,
codebase: if let Some(use_tar) = r.use_tar {
let sh = hash.to_string();
if use_tar {
Some(format!("{sh}.tar"))
.and_then(|r| {
Ok(RawScript {
content: r.content,
lock: r.lock,
meta: Some(ScriptMetadata {
language: r.language,
envs: r.envs,
codebase: if let Some(use_tar) = r.use_tar {
let sh = hash.to_string();
if use_tar {
Some(format!("{sh}.tar"))
} else {
Some(sh)
}
} else {
Some(sh)
}
} else {
None
},
}),
None
},
schema_validator: if r.schema_validation {
r.schema
.as_ref()
.map(|schema_str| {
SchemaValidator::from_schema(schema_str).map_err(|e| anyhow!("Couldn't create schema validator for script requiring schema validation: {e}"))
})
.transpose()?
} else {
None
},
schema: r.schema,
}),
})
})
});
fut.map_ok(|ScriptFull { data, meta }| (data, meta))
@@ -939,6 +960,7 @@ const _: () = {
(u64, |x| format!("{:016x}", x)),
(Uuid, |x| format!("{:032x}", x.as_u128())),
(ScriptHash, |x| format!("{:016x}", x.0)),
((u8, ScriptHash), |x| format!("{:02x}-{:016x}", x.0, x.1.0)),
(FlowNodeId, |x| format!("{:016x}", x.0)),
(AppScriptId, |x| format!("{:016x}", x.0))
}

2
backend/windmill-common/src/error.rs
View File

@@ -74,6 +74,8 @@ pub enum Error {
AlreadyCompleted(String),
#[error("Find python error: {0}")]
FindPythonError(String),
#[error("Problem with arguments: {0}")]
ArgumentErr(String),
}
fn prettify_location(location: &'static Location<'static>) -> String {

1
backend/windmill-common/src/lib.rs
View File

@@ -55,6 +55,7 @@ pub mod utils;
pub mod variables;
pub mod worker;
pub mod workspaces;
pub mod schema;
pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;
pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 5;

653
backend/windmill-common/src/schema.rs Normal file
View File

@@ -0,0 +1,653 @@
use anyhow::anyhow;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use std::{collections::HashMap, str::FromStr};
use serde_json::{value::RawValue, Value};
use crate::{error::Error, scripts::ScriptLang};
#[derive(PartialEq, Serialize, Deserialize, Debug, Clone)]
pub enum JsonPrimitiveType {
String,
Number,
Integer,
Object,
Array,
Boolean,
Null,
}
#[derive(PartialEq, Serialize, Deserialize, Debug, Clone)]
pub enum SchemaValidationRule {
StrictEnum(Vec<Value>),
IsNull,
IsInteger,
IsString,
IsBool,
IsDatetime,
IsNumber,
IsEmail,
IsObject(Vec<(String, Vec<SchemaValidationRule>)>),
IsArray(Vec<SchemaValidationRule>),
IsUnionType(Vec<Vec<SchemaValidationRule>>),
IsOneOf(HashMap<String, Vec<SchemaValidationRule>>),
IsBytes,
}
impl SchemaValidationRule {
fn from_primitive(p: &JsonPrimitiveType, val: &Value) -> Result<Vec<Self>, anyhow::Error> {
let mut schema_rules = vec![];
match p {
JsonPrimitiveType::String => {
schema_rules.push(SchemaValidationRule::IsString);
if let Some(format) = val.get("format").and_then(|f| f.as_str()) {
if format == "date" || format == "date-time" {
schema_rules.push(SchemaValidationRule::IsDatetime);
}
if format == "email" {
schema_rules.push(SchemaValidationRule::IsEmail);
}
}
if let Some(encoding) = val.get("contentEncoding").and_then(|e| e.as_str()) {
if encoding == "base64" {
schema_rules.push(SchemaValidationRule::IsBytes);
}
}
}
JsonPrimitiveType::Number => {
schema_rules.push(SchemaValidationRule::IsNumber);
}
JsonPrimitiveType::Integer => {
schema_rules.push(SchemaValidationRule::IsInteger);
}
JsonPrimitiveType::Object => {
let mut obj_rules = vec![];
if let Some(properties) = val.get("properties") {
let properties = properties
.as_object()
.ok_or(anyhow!("Field properties should be an object"))?;
for (key, v) in properties {
obj_rules.push((key.clone(), SchemaValidationRule::from_value(v)?))
}
schema_rules.push(SchemaValidationRule::IsObject(obj_rules));
} else if let Some(one_of) = val.get("oneOf") {
let one_of = one_of
.as_array()
.ok_or(anyhow!("`oneOf` needs to be an array"))?;
let mut rules_map: HashMap<String, Vec<SchemaValidationRule>> = HashMap::new();
for variant in one_of {
let variant_label = variant
.get("title")
.ok_or(anyhow!(
"oneOf variant definition should have a `title` field"
))?
.as_str()
.ok_or(anyhow!(
"oneOf variant definition `title` field should be a string"
))?;
if !rules_map.contains_key(variant_label) {
rules_map.insert(
variant_label.to_string(),
SchemaValidationRule::from_value(variant)?,
);
} else {
return Err(anyhow!(
"oneOf definition has a duplicate variant `{variant_label}`"
));
}
}
schema_rules.push(SchemaValidationRule::IsOneOf(rules_map))
} else {
let is_resource = val
.get("format")
.and_then(|f| f.as_str())
.map(|f| f.starts_with("resource"))
.unwrap_or(false);
if !is_resource {
return Err(anyhow!(
"Object type should have a `properties` or `anyOf` field, or be a resource"
));
}
}
}
JsonPrimitiveType::Array => {
let items = val
.get("items")
.ok_or(anyhow!("Array type should have field `items`"))?;
let arr_rules = SchemaValidationRule::from_value(items)?;
schema_rules.push(SchemaValidationRule::IsArray(arr_rules));
}
JsonPrimitiveType::Boolean => {
schema_rules.push(SchemaValidationRule::IsBool);
}
JsonPrimitiveType::Null => {
schema_rules.push(SchemaValidationRule::IsNull);
}
}
Ok(schema_rules)
}
fn from_value(val: &Value) -> Result<Vec<Self>, Error> {
if let Some(any_of) = val.get("anyOf").and_then(|any_of| any_of.as_array()) {
let mut r = vec![];
for variant in any_of {
r.push(SchemaValidationRule::from_value(variant)?);
}
return Ok(vec![SchemaValidationRule::IsUnionType(r)]);
}
let mut schema_rules = vec![];
let typ = val.get("type").ok_or(anyhow!("Missing `type` field"))?;
if let Some(typ) = typ.as_str() {
schema_rules.append(&mut SchemaValidationRule::from_primitive(
&JsonPrimitiveType::from_str(typ)?,
val,
)?);
} else if let Some(typ_arr) = typ.as_array() {
let typ_arr = typ_arr
.into_iter()
.map(|v| {
SchemaValidationRule::from_primitive(
&JsonPrimitiveType::from_str(
v.as_str()
.ok_or(anyhow!("Expected array of strings for `type` field"))?,
)?,
v,
)
})
.collect::<Result<Vec<Vec<SchemaValidationRule>>, anyhow::Error>>()?;
schema_rules.push(SchemaValidationRule::IsUnionType(typ_arr));
} else {
return Err(anyhow!(
"Unsupported value for type field, expected string or string array"
)
.into());
}
if let Some(enum_variants) = val.get("enum") {
let variants = enum_variants
.as_array()
.ok_or(anyhow!("enum variants are not in an array"))?
.clone();
schema_rules.push(SchemaValidationRule::StrictEnum(variants));
}
Ok(schema_rules)
}
fn apply_rule(&self, key: &str, val: &Value, required: bool) -> Result<(), Error> {
if val.is_null() {
if !required {
return Ok(());
}
return Err(Error::ArgumentErr(format!("Argument {key} cannot be null")));
}
match self {
SchemaValidationRule::IsNull => {
if !val.is_null() {
return Err(Error::ArgumentErr(format!(
"Argument `{key}` should be null"
)));
}
}
SchemaValidationRule::StrictEnum(vec) => {
if !vec.contains(val) {
let options = vec.iter().map(|s| s.to_string()).join(", ");
return Err(Error::ArgumentErr(format!(
"Enum type argument `{key}` expected one of `[{options}]` but received {}",
val.to_string()
)));
}
}
SchemaValidationRule::IsNumber => {
if !val.is_number() {
return Err(Error::ArgumentErr(format!(
"Argument `{key}` should be a numeric value"
)));
}
}
SchemaValidationRule::IsInteger => {
if !val.is_i64() && !val.is_u64() {
return Err(Error::ArgumentErr(format!(
"Argument `{key}` should be an integer"
)));
}
}
SchemaValidationRule::IsString => {
if !val.is_string() {
return Err(Error::ArgumentErr(format!(
"Argument `{key}` should be a string"
)));
}
}
SchemaValidationRule::IsBool => {
if !val.is_boolean() {
return Err(Error::ArgumentErr(format!(
"Argument `{key}` should be a boolean"
)));
}
}
SchemaValidationRule::IsObject(o) => {
if !val.is_object() {
return Err(Error::ArgumentErr(format!(
"Argument `{key}` should be an object"
)));
}
for (s, rules) in o {
let v = val
.get(&s)
.ok_or(Error::ArgumentErr(format!("Missing field {s} in {key}")))?;
for r in rules {
r.apply_rule(&format!("{key}.{s}"), v, true)?;
}
}
}
SchemaValidationRule::IsArray(vec) => {
if let Some(arr) = val.as_array() {
for (i, el) in arr.iter().enumerate() {
for r in vec {
r.apply_rule(&format!("{key}[{i}]"), el, true)?;
}
}
} else {
return Err(Error::ArgumentErr(format!(
"Argument `{key}` should be an array"
)));
}
}
// TODO: For better error messages on OneOf, make a dedicated OneOf type that matches the label instead of trying the whole type.
SchemaValidationRule::IsUnionType(vec) => {
let mut match_count = 0;
let mut errors = String::new();
for typ in vec {
if let Some(e) = typ
.iter()
.map(|r| r.apply_rule(key, val, true))
.find_map(Result::err)
{
errors.push_str(&format!("- {e}\n"));
} else {
match_count += 1;
}
}
if match_count == 0 {
return Err(Error::ArgumentErr(format!(
"Argument `{key}` is not valid, failed matching to one of the expected types. Here is a list of possible errors:\n{errors}"
)));
}
}
SchemaValidationRule::IsOneOf(vec) => {
let variant_label = val
.get("label")
.ok_or(Error::ArgumentErr(format!(
"oneOf Variant for argument `{key}` should have a label field"
)))?
.as_str()
.ok_or(Error::ArgumentErr(format!(
"Argument `{key}` of type oneOf expected the label to be a string"
)))?;
let variant_rules = vec
.get(variant_label)
.ok_or_else(|| Error::ArgumentErr(format!(
"Argument `{key}` of type oneOf expected one of the following variants {}, but received `{variant_label}`", vec.keys().join(", ")
)))?;
for r in variant_rules {
r.apply_rule(key, val, true).map_err(|e| Error::ArgumentErr(format!("Argument `{key}`: The schema for the selected oneOf variant `{variant_label}` was not respected: {e}")))?;
}
}
// TODO: Implement validation on these
SchemaValidationRule::IsDatetime => (),
SchemaValidationRule::IsEmail => (),
SchemaValidationRule::IsBytes => (),
}
Ok(())
}
}
fn find_annotation(comm_lit: &str, annotation: &str, code: &str) -> bool {
let a = format!("{comm_lit} {annotation}");
for l in code.lines() {
if !l.starts_with(comm_lit) {
break;
}
if l.trim_end() == a {
return true;
}
}
false
}
pub fn should_validate_schema(code: &str, lang: &ScriptLang) -> bool {
let annotation = "schema_validation";
match lang {
ScriptLang::Nativets | ScriptLang::Bun | ScriptLang::Bunnative | ScriptLang::Deno => {
find_annotation("//", annotation, code)
}
ScriptLang::Python3 => find_annotation("#", annotation, code),
ScriptLang::Go => find_annotation("#", annotation, code),
ScriptLang::Bash => find_annotation("#", annotation, code),
ScriptLang::Powershell => find_annotation("#", annotation, code),
ScriptLang::Postgresql => find_annotation("--", annotation, code),
ScriptLang::Mysql => find_annotation("--", annotation, code),
ScriptLang::Bigquery => find_annotation("--", annotation, code),
ScriptLang::Snowflake => find_annotation("--", annotation, code),
ScriptLang::Graphql => find_annotation("#", annotation, code),
ScriptLang::Mssql => find_annotation("--", annotation, code),
ScriptLang::OracleDB => find_annotation("--", annotation, code),
ScriptLang::Php => find_annotation("//", annotation, code),
ScriptLang::Rust => find_annotation("//!", annotation, code),
ScriptLang::Ansible => find_annotation("#", annotation, code),
ScriptLang::CSharp => find_annotation("//", annotation, code),
}
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SchemaValidator {
pub required: Vec<String>,
pub rules: Vec<(String, Vec<SchemaValidationRule>)>,
}
impl SchemaValidator {
pub fn validate(&self, args: &HashMap<String, Box<RawValue>>) -> Result<(), Error> {
for key in &self.required {
if !args.contains_key(key) {
return Err(Error::ArgumentErr(format!("Argument {key} is required")));
}
}
for (key, rules) in &self.rules {
if let Some(raw_val) = args.get(key) {
let parsed_val = Value::from_str(raw_val.get()).map_err(|e| {
Error::ArgumentErr(format!("Failed to parse `{key}` argument: {e}"))
})?;
for rule in rules {
rule.apply_rule(key, &parsed_val, self.required.contains(key))?;
}
}
}
Ok(())
}
pub fn from_schema(schema: &str) -> Result<Self, Error> {
let schema: Value = serde_json::from_str(schema)?;
if let Some(draft_version) = schema.get("$schema") {
match draft_version.as_str() {
Some("https://json-schema.org/draft/2020-12/schema") => (),
_ => return Err(anyhow!("Supplied schema draft version is unsuported").into()),
}
} else {
return Err(anyhow!("No draft version supplied").into());
}
let required: Vec<String> = schema
.get("required")
.ok_or(anyhow!("Missing `required` field on schema"))?
.as_array()
.ok_or(anyhow!("`required` field should be an array of strings"))?
.into_iter()
.map(|v| {
v.as_str()
.map(|s| s.to_string())
.ok_or(anyhow!("required field key is not a string"))
})
.collect::<Result<Vec<String>, anyhow::Error>>()?;
let properties = schema
.get("properties")
.ok_or(anyhow!("Missing `properties` field on schema"))?
.as_object()
.ok_or(anyhow!("`properties` field should be an object"))?;
let mut rules = vec![];
for (key, val) in properties {
rules.push((
key.clone(),
SchemaValidationRule::from_value(val)
.map_err(|e| anyhow!("Problem making rule for {key}: {e}"))?,
));
}
Ok(Self { required, rules })
}
}
impl JsonPrimitiveType {
fn from_str(typ: &str) -> Result<Self, anyhow::Error> {
match typ {
"string" => {
return Ok(JsonPrimitiveType::String);
}
"number" => {
return Ok(JsonPrimitiveType::Number);
}
"integer" => {
return Ok(JsonPrimitiveType::Integer);
}
"object" => {
return Ok(JsonPrimitiveType::Object);
}
"array" => {
return Ok(JsonPrimitiveType::Array);
}
"boolean" => {
return Ok(JsonPrimitiveType::Boolean);
}
"null" => {
return Ok(JsonPrimitiveType::Null);
}
other => return Err(anyhow!("Received unsupported type `{other}`").into()),
}
}
}
#[cfg(test)]
mod tests {
use serde_json::json;
use super::*;
fn value_to_rawvalue_map(
value: Value,
) -> Result<HashMap<String, Box<RawValue>>, anyhow::Error> {
match value {
Value::Object(map) => {
let mut result = HashMap::new();
for (key, val) in map {
let raw = serde_json::to_string(&val)?; // Serialize the Value to a string
let raw_value: Box<RawValue> = serde_json::from_str(&raw)?; // Convert string to Box<RawValue>
result.insert(key, raw_value);
}
Ok(result)
}
_ => Err(anyhow!("Expected a JSON object")),
}
}
#[test]
fn test_parse_and_validate_schema() {
let schema = r#"{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"properties": {
"a": {
"contentEncoding": "base64",
"default": null,
"description": "",
"originalType": "bytes",
"type": "string"
},
"b": {
"default": null,
"description": "",
"enum": [
"my",
"enum"
],
"originalType": "enum",
"type": "string"
},
"e": {
"default": "inferred type string from default arg",
"description": "",
"originalType": "string",
"type": "string"
},
"f": {
"default": {
"nested": "object"
},
"description": "",
"properties": {
"nested": {
"description": "",
"type": "string",
"originalType": "string"
}
},
"type": "object"
},
"g": {
"default": null,
"description": "",
"oneOf": [
{
"type": "object",
"title": "Variant 1",
"properties": {
"label": {
"description": "",
"type": "string",
"originalType": "enum",
"enum": [
"Variant 1"
]
},
"foo": {
"description": "",
"type": "string",
"originalType": "string"
}
}
},
{
"type": "object",
"title": "Variant 2",
"properties": {
"label": {
"description": "",
"type": "string",
"originalType": "enum",
"enum": [
"Variant 2"
]
},
"bar": {
"description": "",
"type": "number"
}
}
}
],
"type": "object"
}
},
"required": [
"a",
"b",
"g"
],
"type": "object"
}
"#;
let validator = SchemaValidator::from_schema(schema)
.expect("Schema couldn't be built from a valid schema");
let args = json!(
{
"g": {
"label": "Variant 1",
"foo": ""
},
"f": {
"nested": "object"
},
"e": "inferred type string from default arg",
"b": "my",
"a": null
}
);
validator
.validate(&value_to_rawvalue_map(args).unwrap())
.err()
.expect("Validation should not work for this");
let args = json!(
{
"g": {
"label": "Variant 1",
"foo": ""
},
"f": {
"nested": "object"
},
"e": "inferred type string from default arg",
"b": "not_enum",
"a": "123"
}
);
validator
.validate(&value_to_rawvalue_map(args).unwrap())
.err()
.expect("Validation should not work for this");
let args = json!(
{
"g": {
"label": "Variant 1",
"foo": ""
},
"f": {
"nested": "object"
},
"e": "inferred type string from default arg",
"b": "my",
"a": "123"
}
);
validator
.validate(&value_to_rawvalue_map(args).unwrap())
.expect("Validation should work for this");
}
}

1
backend/windmill-worker/src/lib.rs
View File

@@ -41,6 +41,7 @@ mod rust_executor;
mod worker;
mod worker_flow;
mod worker_lockfiles;
mod schema;
pub use worker::*;

94
backend/windmill-worker/src/schema.rs Normal file
View File

@@ -0,0 +1,94 @@
use std::collections::HashMap;
use windmill_common::schema::{SchemaValidationRule, SchemaValidator};
use windmill_parser::{MainArgSignature, Typ};
fn make_rules_for_arg_typ(typ: &Typ) -> Vec<SchemaValidationRule> {
let mut rules = vec![];
match typ {
Typ::Str(enum_variants) => {
rules.push(SchemaValidationRule::IsString);
if let Some(enum_variants) = enum_variants {
rules.push(SchemaValidationRule::StrictEnum(
enum_variants
.iter()
.map(|v| serde_json::Value::String(v.to_string()))
.collect(),
));
}
}
Typ::Int => {
rules.push(SchemaValidationRule::IsInteger);
}
Typ::Float => {
rules.push(SchemaValidationRule::IsNumber);
}
Typ::Bool => {
rules.push(SchemaValidationRule::IsBool);
}
Typ::List(typ) => {
rules.push(SchemaValidationRule::IsArray(make_rules_for_arg_typ(typ)));
}
Typ::Bytes => {
rules.push(SchemaValidationRule::IsString);
rules.push(SchemaValidationRule::IsBytes);
}
Typ::Datetime => {
rules.push(SchemaValidationRule::IsString);
rules.push(SchemaValidationRule::IsDatetime);
}
Typ::Email => {
rules.push(SchemaValidationRule::IsString);
rules.push(SchemaValidationRule::IsEmail);
}
Typ::Sql => {
rules.push(SchemaValidationRule::IsString);
}
Typ::Object(props) => {
let mut obj_rules = vec![];
for prop in props {
obj_rules.push((prop.key.to_string(), make_rules_for_arg_typ(&prop.typ)));
}
rules.push(SchemaValidationRule::IsObject(obj_rules))
}
Typ::OneOf(variants) => {
let mut rules_map = HashMap::new();
for variant in variants {
let mut obj_rules = vec![];
for prop in &variant.properties {
obj_rules.push((prop.key.to_string(), make_rules_for_arg_typ(&prop.typ)));
}
rules_map.insert(variant.label.to_string(), vec![SchemaValidationRule::IsObject(obj_rules)]);
}
rules.push(SchemaValidationRule::IsOneOf(rules_map))
}
Typ::Resource(_) => (),
Typ::DynSelect(_) => (),
Typ::Unknown => (),
}
rules
}
pub fn schema_validator_from_main_arg_sig(sig: &MainArgSignature) -> SchemaValidator {
let mut rules = vec![];
let mut required = vec![];
for arg in &sig.args {
if !arg.has_default {
required.push(arg.name.to_string());
}
rules.push((arg.name.to_string(), make_rules_for_arg_typ(&arg.typ)));
}
SchemaValidator { required, rules }
}

204
backend/windmill-worker/src/worker.rs
View File

@@ -9,11 +9,14 @@
// #[cfg(feature = "otel")]
// use opentelemetry::{global, KeyValue};
use anyhow::anyhow;
use futures::TryFutureExt;
use windmill_common::{
apps::AppScriptId,
auth::{fetch_authed_from_permissioned_as, JWTAuthClaims, JobPerms},
cache::{ScriptData, ScriptMetadata},
cache::{future::FutureCachedExt, ScriptData, ScriptMetadata},
jwt,
schema::{should_validate_schema, SchemaValidator},
scripts::PREVIEW_IS_TAR_CODEBASE_HASH,
utils::WarnAfterExt,
worker::{
@@ -49,6 +52,7 @@ use std::{
},
time::Duration,
};
use windmill_parser::MainArgSignature;
use uuid::Uuid;
@@ -108,6 +112,7 @@ use crate::{
js_eval::{eval_fetch_timeout, transpile_ts},
pg_executor::do_postgresql,
result_processor::{process_result, start_background_processor},
schema::schema_validator_from_main_arg_sig,
worker_flow::{handle_flow, update_flow_status_in_progress},
worker_lockfiles::{
handle_app_dependency_job, handle_dependency_job, handle_flow_dependency_job,
@@ -2317,6 +2322,7 @@ pub struct ContentReqLangEnvs {
pub language: Option<ScriptLang>,
pub envs: Option<Vec<String>>,
pub codebase: Option<String>,
pub schema: Option<String>,
}
pub async fn get_hub_script_content_and_requirements(
@@ -2335,6 +2341,7 @@ pub async fn get_hub_script_content_and_requirements(
language: Some(script.language),
envs: None,
codebase: None,
schema: Some(script.schema.get().to_string()),
})
}
@@ -2354,9 +2361,100 @@ pub async fn get_script_content_by_hash(
Some(x) if x.ends_with(".tar") => Some(format!("{}.tar", script_hash)),
Some(_) => Some(script_hash.to_string()),
},
schema: None,
})
}
async fn try_validate_schema(
job: &MiniPulledJob,
db: &Pool<Postgres>,
schema_validator: Option<&SchemaValidator>,
code: &str,
language: Option<&ScriptLang>,
schema: Option<&String>,
) -> Result<(), Error> {
if let Some(args) = job.args.as_ref() {
if let Some(sv) = schema_validator {
sv.validate(args)?;
} else {
let validators_cache = cache::anon!({ (u8, ScriptHash) => Arc<Option<SchemaValidator>> } in "schemavalidators" <= 1000);
let sv_fut = async move {
if language.map(|l| should_validate_schema(code, l)).unwrap_or(false) {
if let Some(schema) = schema {
Ok(Some(SchemaValidator::from_schema(schema)?))
} else {
if let Some(sig) = parse_sig_of_lang(
code,
language,
job.script_entrypoint_override.clone(),
)? {
Ok(Some(schema_validator_from_main_arg_sig(&sig)))
} else {
Err(anyhow!("Job was expected to validate the arguments schema, but no schema was provided and couldn't be inferred from the script for language `{language:?}`. Try removing schema validation for this job").into())
}
}
} else { Ok(None) }
}
.map_ok(Arc::new);
let sub_key: u8 = match job.kind {
JobKind::Script => 0,
JobKind::FlowScript => 1,
JobKind::AppScript => 2,
JobKind::Script_Hub => 3,
JobKind::Preview => 4,
JobKind::DeploymentCallback => 5,
JobKind::SingleScriptFlow => 6,
JobKind::Dependencies => 7,
JobKind::Flow => 8,
JobKind::FlowPreview => 9,
JobKind::Identity => 10,
JobKind::FlowDependencies => 11,
JobKind::AppDependencies => 12,
JobKind::Noop => 13,
JobKind::FlowNode => 14,
};
let sv = match job.runnable_id {
Some(hash)
if job.kind != JobKind::Preview && job.kind != JobKind::FlowPreview =>
{
sv_fut.cached(validators_cache, (sub_key, hash)).await?
}
_ => sv_fut.await?,
};
if sv.is_some() && job.kind == JobKind::Preview {
append_logs(
&job.id,
&job.workspace_id,
"\n--- ARGS VALIDATION ---\nScript contains `schema_validation` annotation, running schema validation for the script arguments...\n",
db,
)
.await;
}
sv.as_ref()
.as_ref()
.map(|sv| sv.validate(args))
.transpose()?;
if sv.is_some() {
append_logs(
&job.id,
&job.workspace_id,
"Script arguments were validated!\n\n",
db,
)
.await;
}
}
}
Ok(())
}
#[tracing::instrument(level = "trace", skip_all)]
async fn handle_code_execution_job(
job: &MiniPulledJob,
@@ -2384,7 +2482,10 @@ async fn handle_code_execution_job(
ScriptData,
ScriptMetadata,
);
let (ScriptData { code, lock }, ScriptMetadata { language, envs, codebase }) = match job.kind {
let (
ScriptData { code, lock },
ScriptMetadata { language, envs, codebase, schema_validator, schema },
) = match job.kind {
JobKind::Preview => {
let codebase = match job.runnable_id.map(|x| x.0) {
Some(PREVIEW_IS_CODEBASE_HASH) => Some(job.id.to_string()),
@@ -2394,15 +2495,22 @@ async fn handle_code_execution_job(
arc_data =
preview.ok_or_else(|| Error::internal_err("expected preview".to_string()))?;
metadata = ScriptMetadata { language: job.script_lang, codebase, envs: None };
metadata = ScriptMetadata {
language: job.script_lang,
codebase,
envs: None,
schema: None,
schema_validator: None,
};
(arc_data.as_ref(), &metadata)
}
JobKind::Script_Hub => {
let ContentReqLangEnvs { content, lockfile, language, envs, codebase } =
let ContentReqLangEnvs { content, lockfile, language, envs, codebase, schema } =
get_hub_script_content_and_requirements(job.runnable_path.as_ref(), Some(db))
.await?;
data = ScriptData { code: content, lock: lockfile };
metadata = ScriptMetadata { language, envs, codebase };
metadata = ScriptMetadata { language, envs, codebase, schema, schema_validator: None };
(&data, &metadata)
}
JobKind::Script => {
@@ -2411,12 +2519,24 @@ async fn handle_code_execution_job(
}
JobKind::FlowScript => {
arc_data = cache::flow::fetch_script(db, FlowNodeId(script_hash()?.0)).await?;
metadata = ScriptMetadata { language: job.script_lang, envs: None, codebase: None };
metadata = ScriptMetadata {
language: job.script_lang,
envs: None,
codebase: None,
schema: None,
schema_validator: None,
};
(arc_data.as_ref(), &metadata)
}
JobKind::AppScript => {
arc_data = cache::app::fetch_script(db, AppScriptId(script_hash()?.0)).await?;
metadata = ScriptMetadata { language: job.script_lang, envs: None, codebase: None };
metadata = ScriptMetadata {
language: job.script_lang,
envs: None,
codebase: None,
schema: None,
schema_validator: None,
};
(arc_data.as_ref(), &metadata)
}
JobKind::DeploymentCallback => {
@@ -2425,10 +2545,11 @@ async fn handle_code_execution_job(
.as_ref()
.ok_or_else(|| Error::internal_err("expected script path".to_string()))?;
if script_path.starts_with("hub/") {
let ContentReqLangEnvs { content, lockfile, language, envs, codebase } =
let ContentReqLangEnvs { content, lockfile, language, envs, codebase, schema } =
get_hub_script_content_and_requirements(Some(script_path), Some(db)).await?;
data = ScriptData { code: content, lock: lockfile };
metadata = ScriptMetadata { language, envs, codebase };
metadata =
ScriptMetadata { language, envs, codebase, schema, schema_validator: None };
(&data, &metadata)
} else {
let hash = sqlx::query_scalar!(
@@ -2450,6 +2571,16 @@ async fn handle_code_execution_job(
),
};
try_validate_schema(
job,
db,
schema_validator.as_ref(),
code,
language.as_ref(),
schema.as_ref(),
)
.await?;
let language = language.clone();
if language == Some(ScriptLang::Postgresql) {
return do_postgresql(
@@ -2908,3 +3039,58 @@ mount {{
result
}
fn parse_sig_of_lang(
code: &str,
language: Option<&ScriptLang>,
main_override: Option<String>,
) -> Result<Option<MainArgSignature>> {
Ok(if let Some(lang) = language {
match lang {
ScriptLang::Nativets | ScriptLang::Deno | ScriptLang::Bun | ScriptLang::Bunnative => {
Some(windmill_parser_ts::parse_deno_signature(
code,
true,
false,
main_override,
)?)
}
#[cfg(feature = "python")]
ScriptLang::Python3 => Some(windmill_parser_py::parse_python_signature(
code,
main_override,
false,
)?),
#[cfg(not(feature = "python"))]
ScriptLang::Python3 => None,
ScriptLang::Go => Some(windmill_parser_go::parse_go_sig(code)?),
ScriptLang::Bash => Some(windmill_parser_bash::parse_bash_sig(code)?),
ScriptLang::Powershell => Some(windmill_parser_bash::parse_powershell_sig(code)?),
ScriptLang::Postgresql => Some(windmill_parser_sql::parse_pgsql_sig(code)?),
ScriptLang::Mysql => Some(windmill_parser_sql::parse_mysql_sig(code)?),
ScriptLang::Bigquery => Some(windmill_parser_sql::parse_bigquery_sig(code)?),
ScriptLang::Snowflake => Some(windmill_parser_sql::parse_snowflake_sig(code)?),
ScriptLang::Graphql => None,
ScriptLang::Mssql => Some(windmill_parser_sql::parse_mssql_sig(code)?),
ScriptLang::OracleDB => Some(windmill_parser_sql::parse_oracledb_sig(code)?),
#[cfg(feature = "php")]
ScriptLang::Php => Some(windmill_parser_php::parse_php_signature(
code,
main_override,
)?),
#[cfg(not(feature = "php"))]
ScriptLang::Php => None,
#[cfg(feature = "rust")]
ScriptLang::Rust => Some(windmill_parser_rust::parse_rust_signature(code)?),
#[cfg(not(feature = "rust"))]
ScriptLang::Rust => None,
ScriptLang::Ansible => Some(windmill_parser_yaml::parse_ansible_sig(code)?),
#[cfg(feature = "csharp")]
ScriptLang::CSharp => Some(windmill_parser_csharp::parse_csharp_signature(code)?),
#[cfg(not(feature = "csharp"))]
ScriptLang::CSharp => None,
}
} else {
None
})
}

30
backend/windmill-worker/src/worker_lockfiles.rs
View File

@@ -1,3 +1,4 @@
use std::borrow::Cow;
use std::collections::HashMap;
use std::path::{Component, Path, PathBuf};
@@ -259,13 +260,36 @@ pub async fn handle_dependency_job(
// `JobKind::Dependencies` job store either:
// - A saved script `hash` in the `script_hash` column.
// - Preview raw lock and code in the `queue` or `job` table.
let script_data = match job.runnable_id {
Some(hash) => &cache::script::fetch(db, hash).await?.0,
let script_data = &match job.runnable_id {
Some(hash) => match cache::script::fetch(db, hash).await {
Ok(d) => Cow::Owned(d.0),
Err(e) => {
let logs2 = sqlx::query_scalar!(
"SELECT logs FROM job_logs WHERE job_id = $1 AND workspace_id = $2",
&job.id,
&job.workspace_id
)
.fetch_optional(db)
.await?
.flatten()
.unwrap_or_else(|| "no logs".to_string());
sqlx::query!(
"UPDATE script SET lock_error_logs = $1 WHERE hash = $2 AND workspace_id = $3",
&format!("{logs2}\n{e}"),
&job.runnable_id.unwrap_or(ScriptHash(0)).0,
&job.workspace_id
)
.execute(db)
.await?;
return Err(Error::ExecutionErr(format!("Error creating schema validator: {e}")))
}
},
_ => match preview_data {
Some(RawData::Script(data)) => data,
Some(RawData::Script(data)) => Cow::Borrowed(data),
_ => return Err(Error::internal_err("expected script hash")),
},
};
let content = capture_dependency_job(
&job.id,
job.script_lang.as_ref().map(|v| Ok(v)).unwrap_or_else(|| {