starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-26 08:01:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: prevent native http against timeout abuse

Browse Source
This commit is contained in:
Ruben Fiszel
2024-02-19 23:28:06 +01:00
parent 8e21532684
commit e28fbc9c97
3 changed files with 246 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/common.rs
+1 -1
View File
@@ -487,7 +487,7 @@ where
result = tokio::time::timeout(std::time::Duration::from_millis(timeout_ms), result_f) => result
.map_err(|e| {
tracing::error!("Query timeout: {}", e);
Error::ExecutionErr("Query timeout".to_string())
Error::ExecutionErr(format!("Query timeout after (>{}s)", timeout_ms/1000))
})?,
_ = update_job, if job_id != Uuid::nil() => Err(Error::ExecutionErr("Job cancelled".to_string())).map_err(to_anyhow)?,
}?;
backend/windmill-worker/src/js_eval.rs
+218 -151
View File
@@ -6,11 +6,21 @@
* LICENSE-AGPL for a copy of the license.
*/
use std::{cell::RefCell, collections::HashMap, rc::Rc, sync::Arc, env, io::{BufReader, self}};
use std::{
cell::RefCell,
collections::HashMap,
env,
io::{self, BufReader},
rc::Rc,
sync::Arc,
};
use deno_ast::{ParseParams, SourceTextInfo};
use deno_core::{
error::AnyError, op2, serde_v8, url, v8::{self, IsolateHandle}, Extension, JsRuntime, Op, OpState, PollEventLoopOptions, RuntimeOptions, Snapshot
error::AnyError,
op2, serde_v8, url,
v8::{self, IsolateHandle},
Extension, JsRuntime, Op, OpState, PollEventLoopOptions, RuntimeOptions, Snapshot,
};
use deno_fetch::FetchPermissions;
use deno_tls::{rustls::RootCertStore, rustls_pemfile};
@@ -25,9 +35,13 @@ use tokio::{
time::timeout,
};
use uuid::Uuid;
use windmill_common::{error::Error, flow_status::JobResult};
use windmill_common::{error::Error, flow_status::JobResult, DB};
use windmill_queue::CanceledBy;
use crate::{common::unsafe_raw, AuthedClient};
use crate::{
common::{run_future_with_polling_update_job_poller, unsafe_raw},
AuthedClient,
};
#[derive(Debug, Clone)]
pub struct IdContext {
@@ -44,7 +58,7 @@ impl ContainerRootCertStoreProvider {
fn new() -> ContainerRootCertStoreProvider {
return ContainerRootCertStoreProvider {
root_cert_store: deno_tls::create_default_root_cert_store(),
}
};
}
fn add_certificate(&mut self, cert_path: String) -> io::Result<()> {
@@ -101,40 +115,58 @@ pub async fn eval_timeout(
authed_client: Option<&AuthedClient>,
by_id: Option<IdContext>,
) -> anyhow::Result<Box<RawValue>> {
let expr = expr.trim().to_string();
for (k,v) in transform_context.iter() {
for (k, v) in transform_context.iter() {
if k == &expr {
return Ok(v.as_ref().clone())
return Ok(v.as_ref().clone());
}
}
if expr.starts_with("flow_input.") || expr.starts_with("flow_input[") {
if let Some(ref flow_input) = flow_input {
for (k,v) in flow_input.iter() {
if &format!("flow_input.{k}") == &expr || &format!("flow_input[\"{k}\"]") == &expr {
for (k, v) in flow_input.iter() {
if &format!("flow_input.{k}") == &expr || &format!("flow_input[\"{k}\"]") == &expr {
// tracing::error!("FLOW_INPUT");
return Ok(v.clone())
return Ok(v.clone());
}
}
}
}
let p_id = by_id.as_ref().map(|x| format!("results.{}", x.previous_id));
let p_id2 = by_id.as_ref().map(|x| format!("results[\"{}\"]", x.previous_id));
let p_id2 = by_id
.as_ref()
.map(|x| format!("results[\"{}\"]", x.previous_id));
if p_id.is_some() && transform_context.contains_key("previous_result") && &expr == p_id.as_ref().unwrap() {
if p_id.is_some()
&& transform_context.contains_key("previous_result")
&& &expr == p_id.as_ref().unwrap()
{
// tracing::error!("PREVIOUS_RESULT");
return Ok(transform_context.get("previous_result").unwrap().as_ref().clone())
return Ok(transform_context
.get("previous_result")
.unwrap()
.as_ref()
.clone());
}
if p_id2.is_some() && transform_context.contains_key("previous_result") && &expr == p_id2.as_ref().unwrap() {
return Ok(transform_context.get("previous_result").unwrap().as_ref().clone())
if p_id2.is_some()
&& transform_context.contains_key("previous_result")
&& &expr == p_id2.as_ref().unwrap()
{
return Ok(transform_context
.get("previous_result")
.unwrap()
.as_ref()
.clone());
}
if by_id.is_some() && authed_client.is_some() {
if let Some(x) = RE_FULL.captures(&expr).and_then(|x| x.get(1).map(|y| y.as_str())) {
if let Some(x) = RE_FULL
.captures(&expr)
.and_then(|x| x.get(1).map(|y| y.as_str()))
{
// tracing::error!("{:?}", x.split(".").collect::<Vec<_>>());
let arr = x.split(".").collect::<Vec<_>>();
let mut iter = arr.iter();
@@ -142,12 +174,14 @@ pub async fn eval_timeout(
if let Some(id) = iter.next() {
let path = iter.join(".");
let query = if path.is_empty() { None } else { Some(path) };
return authed_client.unwrap().get_result_by_id(&by_id.as_ref().unwrap().flow_job.to_string(), id, query).await;
return authed_client
.unwrap()
.get_result_by_id(&by_id.as_ref().unwrap().flow_job.to_string(), id, query)
.await;
}
}
}
let expr2 = expr.clone();
let (sender, mut receiver) = oneshot::channel::<IsolateHandle>();
let has_client = authed_client.is_some();
@@ -181,16 +215,17 @@ pub async fn eval_timeout(
..Default::default()
};
let mut context_keys = transform_context
.keys()
.filter(|x|
expr.contains(&x.to_string())
)
.filter(|x| expr.contains(&x.to_string()))
.map(|x| x.clone())
.collect_vec();
if !context_keys.contains(&"previous_result".to_string()) && (p_id.is_some() && expr.contains(p_id.as_ref().unwrap())) || expr.contains("error") || (p_id2.is_some() && expr.contains(p_id2.as_ref().unwrap())) {
if !context_keys.contains(&"previous_result".to_string())
&& (p_id.is_some() && expr.contains(p_id.as_ref().unwrap()))
|| expr.contains("error")
|| (p_id2.is_some() && expr.contains(p_id2.as_ref().unwrap()))
{
context_keys.push("previous_result".to_string());
}
let has_flow_input = expr.contains("flow_input");
@@ -204,10 +239,15 @@ pub async fn eval_timeout(
let mut op_state = op_state.borrow_mut();
let mut client = authed_client.clone();
if let Some(client) = client.as_mut() {
client.force_client = Some(reqwest::ClientBuilder::new()
.user_agent("windmill/beta")
.danger_accept_invalid_certs(std::env::var("ACCEPT_INVALID_CERTS").is_ok())
.build().unwrap());
client.force_client = Some(
reqwest::ClientBuilder::new()
.user_agent("windmill/beta")
.danger_accept_invalid_certs(
std::env::var("ACCEPT_INVALID_CERTS").is_ok(),
)
.build()
.unwrap(),
);
}
op_state.put(OptAuthedClient(client));
op_state.put(TransformContext {
@@ -266,9 +306,11 @@ fn replace_with_await(expr: String, fn_name: &str) -> String {
s
}
lazy_static! {
static ref RE: Regex = Regex::new(r#"(?m)(?P<r>results(?:(?:\.(?:[a-z]|[A-Z]|_|[1-9])+)|(?:\[\".*?\"\])))"#).unwrap();
static ref RE_FULL: Regex = Regex::new(r"(?m)^results((?:\.(?:(?:[a-z]|[A-Z]|_|[1-9])+))+)$").unwrap();
static ref RE: Regex =
Regex::new(r#"(?m)(?P<r>results(?:(?:\.(?:[a-z]|[A-Z]|_|[1-9])+)|(?:\[\".*?\"\])))"#)
.unwrap();
static ref RE_FULL: Regex =
Regex::new(r"(?m)^results((?:\.(?:(?:[a-z]|[A-Z]|_|[1-9])+))+)$").unwrap();
}
fn replace_with_await_result(expr: String) -> String {
@@ -396,10 +438,11 @@ function get_from_env(name) {{
},
);
let script = context.execute_script("<anon>", code.into())?;
let fut = context.resolve(script);
let global = context.with_event_loop_promise(fut, PollEventLoopOptions::default()).await?;
let global = context
.with_event_loop_promise(fut, PollEventLoopOptions::default())
.await?;
let scope = &mut context.handle_scope();
let local = v8::Local::new(scope, global);
@@ -459,12 +502,12 @@ async fn op_get_id(
#[string] flow_job_id: String,
#[string] node_id: String,
) -> Result<Option<String>, anyhow::Error> {
let client = op_state.borrow().borrow::<OptAuthedClient>().0.clone();
if let Some(client) = client {
let result = client
.get_result_by_id::<Option<Box<RawValue>>>(&flow_job_id, &node_id, None)
.await.ok();
.await
.ok();
if let Some(result) = result {
Ok(result.map(|x| x.get().to_string()))
} else {
@@ -479,7 +522,7 @@ async fn op_get_id(
#[serde]
async fn op_resource(
op_state: Rc<RefCell<OpState>>,
#[string] path:String,
#[string] path: String,
) -> Result<serde_json::Value, anyhow::Error> {
let client = op_state.borrow().borrow::<OptAuthedClient>().0.clone();
if let Some(client) = client {
@@ -496,7 +539,7 @@ pub struct TransformContext {
#[op2]
#[string]
fn op_get_context(op_state: Rc<RefCell<OpState>>, #[string] id: &str) -> String {
fn op_get_context(op_state: Rc<RefCell<OpState>>, #[string] id: &str) -> String {
let ops = op_state.borrow();
let client = ops.borrow::<TransformContext>();
if id == "flow_input" {
@@ -506,11 +549,11 @@ fn op_get_context(op_state: Rc<RefCell<OpState>>, #[string] id: &str) -> String
.and_then(|x| serde_json::to_string(&x).ok())
.unwrap_or_else(|| "null".to_string())
} else {
client
.envs
.get(id)
.and_then(|x| serde_json::to_string(x).ok())
.unwrap_or_else(String::new)
client
.envs
.get(id)
.and_then(|x| serde_json::to_string(x).ok())
.unwrap_or_else(String::new)
}
}
@@ -540,121 +583,142 @@ pub async fn eval_fetch_timeout(
ts_expr: String,
js_expr: String,
args: Option<&Json<HashMap<String, Box<RawValue>>>>,
job_id: Uuid,
job_timeout: Option<i32>,
db: &DB,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
worker_name: &str,
w_id: &str,
) -> anyhow::Result<(Box<RawValue>, String)> {
let (sender, mut receiver) = oneshot::channel::<IsolateHandle>();
let ts_expr2 = ts_expr.clone();
let parsed_args = windmill_parser_ts::parse_deno_signature(&ts_expr, true)?.args;
let spread = parsed_args.into_iter().map(|x| args.as_ref().and_then(|args| args.0.get(&x.name).map(|x| x.clone()))).collect::<Vec<_>>();
let spread = parsed_args
.into_iter()
.map(|x| {
args.as_ref()
.and_then(|args| args.0.get(&x.name).map(|x| x.clone()))
})
.collect::<Vec<_>>();
timeout(
std::time::Duration::from_secs(100),
tokio::task::spawn_blocking(move || {
let ops = vec![op_get_static_args::DECL, op_log::DECL];
let ext = Extension {
name: "windmill",
ops: ops.into(),
..Default::default()
};
let result_f = tokio::task::spawn_blocking(move || {
let ops = vec![op_get_static_args::DECL, op_log::DECL];
let ext = Extension { name: "windmill", ops: ops.into(), ..Default::default() };
let deno_fetch_options = if let Some(cert_path) = env::var("DENO_CERT").ok() {
let mut cert_store_provider = ContainerRootCertStoreProvider::new();
cert_store_provider.add_certificate(cert_path)?;
let deno_fetch_options = if let Some(cert_path) = env::var("DENO_CERT").ok() {
let mut cert_store_provider = ContainerRootCertStoreProvider::new();
cert_store_provider.add_certificate(cert_path)?;
deno_fetch::Options {
root_cert_store_provider: Some(Arc::new(cert_store_provider)),
..Default::default()
}
} else {
Default::default()
};
let exts: Vec<Extension> = vec![
deno_webidl::deno_webidl::init_ops(),
deno_url::deno_url::init_ops(),
deno_console::deno_console::init_ops(),
deno_web::deno_web::init_ops::<PermissionsContainer>(
Arc::new(BlobStore::default()),
None,
),
deno_fetch::deno_fetch::init_ops::<PermissionsContainer>(deno_fetch_options),
ext
];
// Use our snapshot to provision our new runtime
let options = RuntimeOptions {
is_main: true,
extensions: exts,
create_params: Some(deno_core::v8::CreateParams::default().heap_limits(
0 as usize,
1024*1024*128 as usize,
)),
startup_snapshot: Some(Snapshot::Static(RUNTIME_SNAPSHOT)),
module_loader: Some(Rc::new(deno_core::FsModuleLoader)),
// startup_snapshot: Some(Snapshot::Static(buffer)),
deno_fetch::Options {
root_cert_store_provider: Some(Arc::new(cert_store_provider)),
..Default::default()
};
let (memory_limit_tx, mut memory_limit_rx) = mpsc::unbounded_channel::<()>();
// tracing::info!("starting isolate");
// let instant = Instant::now();
let mut js_runtime: JsRuntime = JsRuntime::new(options);
// tracing::info!("ttc: {:?}", instant.elapsed());
js_runtime.add_near_heap_limit_callback(move |x,y| {
tracing::error!("heap limit reached: {x} {y}");
if memory_limit_tx.send(()).is_err() {
tracing::error!("failed to send memory limit reached notification - isolate may already be terminating");
};
//to give a bit of time to kill the worker without v8 crashing
return y*2;
});
{
let op_state = js_runtime.op_state();
let mut op_state = op_state.borrow_mut();
op_state.put(PermissionsContainer{});
//reqwest client seems to not be sharable between runtimes unfortunately
// op_state.put(HTTP_CLIENT.clone());
op_state.put(MainArgs { args: spread });
op_state.put(LogString { s: String::new() });
}
} else {
Default::default()
};
let exts: Vec<Extension> = vec![
deno_webidl::deno_webidl::init_ops(),
deno_url::deno_url::init_ops(),
deno_console::deno_console::init_ops(),
deno_web::deno_web::init_ops::<PermissionsContainer>(
Arc::new(BlobStore::default()),
None,
),
deno_fetch::deno_fetch::init_ops::<PermissionsContainer>(deno_fetch_options),
ext,
];
sender
.send(js_runtime.v8_isolate().thread_safe_handle())
.map_err(|_| Error::ExecutionErr("impossible to send v8 isolate".to_string()))?;
// Use our snapshot to provision our new runtime
let options = RuntimeOptions {
is_main: true,
extensions: exts,
create_params: Some(
deno_core::v8::CreateParams::default()
.heap_limits(0 as usize, 1024 * 1024 * 128 as usize),
),
startup_snapshot: Some(Snapshot::Static(RUNTIME_SNAPSHOT)),
module_loader: Some(Rc::new(deno_core::FsModuleLoader)),
// startup_snapshot: Some(Snapshot::Static(buffer)),
..Default::default()
};
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let (memory_limit_tx, mut memory_limit_rx) = mpsc::unbounded_channel::<()>();
// tracing::info!("starting isolate");
// let instant = Instant::now();
let future = async { tokio::select! {
let mut js_runtime: JsRuntime = JsRuntime::new(options);
// tracing::info!("ttc: {:?}", instant.elapsed());
js_runtime.add_near_heap_limit_callback(move |x,y| {
tracing::error!("heap limit reached: {x} {y}");
if memory_limit_tx.send(()).is_err() {
tracing::error!("failed to send memory limit reached notification - isolate may already be terminating");
};
//to give a bit of time to kill the worker without v8 crashing
return y*2;
});
{
let op_state = js_runtime.op_state();
let mut op_state = op_state.borrow_mut();
op_state.put(PermissionsContainer {});
//reqwest client seems to not be sharable between runtimes unfortunately
// op_state.put(HTTP_CLIENT.clone());
op_state.put(MainArgs { args: spread });
op_state.put(LogString { s: String::new() });
}
sender
.send(js_runtime.v8_isolate().thread_safe_handle())
.map_err(|_| Error::ExecutionErr("impossible to send v8 isolate".to_string()))?;
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let future = async {
tokio::select! {
r = eval_fetch(&mut js_runtime, &js_expr) => Ok(r),
_ = memory_limit_rx.recv() => Err(Error::ExecutionErr("Memory limit reached, killing isolate".to_string()))
}};
let r = runtime.block_on(future)?;
// tracing::info!("total: {:?}", instant.elapsed());
}
};
let r = runtime.block_on(future)?;
// tracing::info!("total: {:?}", instant.elapsed());
(r as anyhow::Result<Box<RawValue>>).map(|x| (x, js_runtime.op_state().borrow().borrow::<LogString>().s.clone()))
}),
(r as anyhow::Result<Box<RawValue>>).map(|x| {
(
x,
js_runtime
.op_state()
.borrow()
.borrow::<LogString>()
.s
.clone(),
)
})
});
run_future_with_polling_update_job_poller(
job_id,
job_timeout,
db,
mem_peak,
canceled_by,
async { result_f.await? },
worker_name,
w_id,
)
.await
.map_err(|_| {
.map_err(|e| {
if let Ok(isolate) = receiver.try_recv() {
isolate.terminate_execution();
};
Error::ExecutionErr(format!(
"The expression of evaluation `{ts_expr2}` took too long to execute (>100s)"
))
})??
}
e
})
}
async fn eval_fetch(js_runtime: &mut JsRuntime, expr: &str) -> anyhow::Result<Box<RawValue>> {
@@ -676,8 +740,9 @@ import("file:///eval.ts").then((module) => module.main(...args)).then(JSON.strin
)?;
let fut = js_runtime.resolve(script);
let global = js_runtime.with_event_loop_promise(fut, PollEventLoopOptions::default()).await?;
let global = js_runtime
.with_event_loop_promise(fut, PollEventLoopOptions::default())
.await?;
let scope = &mut js_runtime.handle_scope();
let local = v8::Local::new(scope, global);
@@ -690,7 +755,13 @@ import("file:///eval.ts").then((module) => module.main(...args)).then(JSON.strin
#[op2]
#[serde]
fn op_get_static_args(op_state: Rc<RefCell<OpState>>) -> Vec<Option<String>> {
op_state.borrow().borrow::<MainArgs>().args.iter().map(|x| x.as_ref().map(|y| y.get().to_string())).collect_vec()
op_state
.borrow()
.borrow::<MainArgs>()
.args
.iter()
.map(|x| x.as_ref().map(|y| y.get().to_string()))
.collect_vec()
}
#[op2(fast)]
@@ -730,11 +801,7 @@ mod tests {
let ext = Extension { name: "js_eval", ops: ops.into(), ..Default::default() };
let exts = vec![ext];
let options = RuntimeOptions {
extensions: exts,
..Default::default()
};
let options = RuntimeOptions { extensions: exts, ..Default::default() };
let mut runtime = JsRuntime::new(options);
{
@@ -794,12 +861,12 @@ multiline template`";
Ok(())
}
#[tokio::test]
async fn test_eval_fetch_timeout() -> anyhow::Result<()> {
let code = r#"export async function main() { return "" }"#;
// #[tokio::test]
// async fn test_eval_fetch_timeout() -> anyhow::Result<()> {
// let code = r#"export async function main() { return "" }"#;
let res = eval_fetch_timeout(code.to_string(), code.to_string(), None).await?;
assert_eq!(res.0.get(), "\"\"");
Ok(())
}
// let res = eval_fetch_timeout(code.to_string(), code.to_string(), None, Uuid::new_v4(), None, ).await?;
// assert_eq!(res.0.get(), "\"\"");
// Ok(())
// }
}
backend/windmill-worker/src/worker.rs
+27 -2
View File
@@ -2374,6 +2374,9 @@ async fn do_nativets(
client: &AuthedClientBackgroundTask,
code: String,
db: &Pool<Postgres>,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
worker_name: &str,
) -> windmill_common::error::Result<(Box<RawValue>, String)> {
let args = build_args_map(job, client, db).await?.map(Json);
let job_args = if args.is_some() {
@@ -2382,7 +2385,19 @@ async fn do_nativets(
job.args.as_ref()
};
let result = eval_fetch_timeout(code.clone(), transpile_ts(code)?, job_args).await?;
let result = eval_fetch_timeout(
code.clone(),
transpile_ts(code)?,
job_args,
job.id,
job.timeout,
db,
mem_peak,
canceled_by,
worker_name,
&job.workspace_id,
)
.await?;
Ok((result.0, [logs, result.1].join("\n\n")))
}
@@ -2948,7 +2963,17 @@ async fn handle_code_execution_job(
&client.get_token().await,
inner_content
);
let (result, ts_logs) = do_nativets(job, logs.clone(), &client, code, db).await?;
let (result, ts_logs) = do_nativets(
job,
logs.clone(),
&client,
code,
db,
mem_peak,
canceled_by,
worker_name,
)
.await?;
*logs = ts_logs;
return Ok(result);
}