diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 45d1d65dd1..8beacf0c61 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -17361,6 +17361,7 @@ dependencies = [ "derive_more 1.0.0", "dotenv", "eventsource-stream", + "flate2", "flume", "futures", "gcp_auth", @@ -17402,6 +17403,7 @@ dependencies = [ "sha2 0.10.9", "sqlx", "tar", + "tempfile", "tiberius", "tokio", "tokio-postgres 0.7.13", diff --git a/backend/migrations/20260217000000_sandbox_snapshots_volumes.down.sql b/backend/migrations/20260217000000_sandbox_snapshots_volumes.down.sql new file mode 100644 index 0000000000..489701e489 --- /dev/null +++ b/backend/migrations/20260217000000_sandbox_snapshots_volumes.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS sandbox_volume; +DROP TABLE IF EXISTS sandbox_snapshot; diff --git a/backend/migrations/20260217000000_sandbox_snapshots_volumes.up.sql b/backend/migrations/20260217000000_sandbox_snapshots_volumes.up.sql new file mode 100644 index 0000000000..aca8eb7de5 --- /dev/null +++ b/backend/migrations/20260217000000_sandbox_snapshots_volumes.up.sql @@ -0,0 +1,32 @@ +-- Add sandbox_snapshot and sandbox_volume tables + +CREATE TABLE sandbox_snapshot ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id), + name VARCHAR(255) NOT NULL, + tag VARCHAR(255) NOT NULL DEFAULT 'latest', + s3_key TEXT NOT NULL, + content_hash VARCHAR(64) NOT NULL DEFAULT '', + docker_image TEXT NOT NULL, + setup_script TEXT, + size_bytes BIGINT, + status VARCHAR(20) NOT NULL DEFAULT 'pending', + build_error TEXT, + build_job_id UUID, + created_by VARCHAR(255) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + extra_perms JSONB NOT NULL DEFAULT '{}'::jsonb, + PRIMARY KEY (workspace_id, name, tag) +); + +CREATE TABLE sandbox_volume ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id), + name VARCHAR(255) NOT NULL, + s3_key TEXT NOT NULL, + size_bytes BIGINT, + created_by VARCHAR(255) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + extra_perms JSONB NOT NULL DEFAULT '{}'::jsonb, + PRIMARY KEY (workspace_id, name) +); diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 76fe0224ed..b8093205ec 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -131,6 +131,7 @@ pub mod oidc_ee; mod oidc_oss; mod raw_apps; mod resources; +mod sandbox; #[cfg(feature = "private")] pub mod saml_ee; mod saml_oss; @@ -509,6 +510,7 @@ pub async fn run_server( .nest("/npm_proxy", windmill_api_npm_proxy::workspaced_service()) .nest("/raw_apps", raw_apps::workspaced_service()) .nest("/resources", resources::workspaced_service()) + .nest("/sandbox", sandbox::workspaced_service()) .nest("/schedules", windmill_api_schedule::workspaced_service()) .nest("/scripts", scripts::workspaced_service()) .nest( diff --git a/backend/windmill-api/src/sandbox.rs b/backend/windmill-api/src/sandbox.rs new file mode 100644 index 0000000000..0edd873d20 --- /dev/null +++ b/backend/windmill-api/src/sandbox.rs @@ -0,0 +1,303 @@ +use axum::{ + extract::{Extension, Path, Query}, + routing::{delete, get, post}, + Json, Router, +}; +use serde::Deserialize; +use windmill_common::{ + db::DB, + error::{self, JsonResult}, + sandbox::{SandboxSnapshot, SandboxVolume}, + utils::require_admin, +}; + +use crate::db::ApiAuthed; + +#[cfg(feature = "parquet")] +async fn delete_s3_object(s3_key: &str) { + if let Some(os) = windmill_common::s3_helpers::get_object_store().await { + let path = object_store::path::Path::from(s3_key); + if let Err(e) = os.delete(&path).await { + tracing::warn!("Failed to delete S3 object {s3_key}: {e}"); + } + } +} + +#[cfg(not(feature = "parquet"))] +async fn delete_s3_object(_s3_key: &str) {} + +pub fn workspaced_service() -> Router { + Router::new() + .route("/snapshots", get(list_snapshots).post(create_snapshot)) + .route("/snapshots/:name/:tag", get(get_snapshot).delete(delete_snapshot)) + .route("/snapshots/:name", delete(delete_snapshot_all_tags)) + .route("/snapshots/:name/:tag/rebuild", post(rebuild_snapshot)) + .route("/volumes", get(list_volumes).post(create_volume)) + .route("/volumes/:name", get(get_volume).delete(delete_volume)) +} + +#[derive(Deserialize)] +struct ListSnapshotsQuery { + name: Option, +} + +async fn list_snapshots( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + Query(query): Query, +) -> JsonResult> { + require_admin(authed.is_admin, &authed.username)?; + let rows = if let Some(name) = query.name { + sqlx::query_as!( + SandboxSnapshot, + "SELECT workspace_id, name, tag, s3_key, content_hash, docker_image, setup_script, \ + size_bytes, status, build_error, build_job_id, created_by, created_at, updated_at, \ + extra_perms \ + FROM sandbox_snapshot WHERE workspace_id = $1 AND name = $2 ORDER BY created_at DESC", + &w_id, + &name, + ) + .fetch_all(&db) + .await? + } else { + sqlx::query_as!( + SandboxSnapshot, + "SELECT workspace_id, name, tag, s3_key, content_hash, docker_image, setup_script, \ + size_bytes, status, build_error, build_job_id, created_by, created_at, updated_at, \ + extra_perms \ + FROM sandbox_snapshot WHERE workspace_id = $1 ORDER BY created_at DESC", + &w_id, + ) + .fetch_all(&db) + .await? + }; + Ok(Json(rows)) +} + +#[derive(Deserialize)] +struct CreateSnapshot { + name: String, + #[serde(default = "default_tag")] + tag: String, + docker_image: String, + setup_script: Option, +} + +fn default_tag() -> String { + "latest".to_string() +} + +async fn create_snapshot( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + Json(body): Json, +) -> error::Result { + require_admin(authed.is_admin, &authed.username)?; + let s3_key = format!( + "sandbox/snapshots/{}/{}/{}.tar.gz", + w_id, body.name, body.tag + ); + sqlx::query!( + "INSERT INTO sandbox_snapshot \ + (workspace_id, name, tag, s3_key, docker_image, setup_script, created_by, status) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, 'pending') \ + ON CONFLICT (workspace_id, name, tag) DO UPDATE SET \ + docker_image = $5, setup_script = $6, status = 'pending', \ + build_error = NULL, updated_at = now()", + &w_id, + &body.name, + &body.tag, + &s3_key, + &body.docker_image, + body.setup_script.as_deref(), + &authed.username, + ) + .execute(&db) + .await?; + + Ok(format!( + "Snapshot {}:{} created with status 'pending'", + body.name, body.tag + )) +} + +async fn get_snapshot( + authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, name, tag)): Path<(String, String, String)>, +) -> JsonResult { + require_admin(authed.is_admin, &authed.username)?; + let row = sqlx::query_as!( + SandboxSnapshot, + "SELECT workspace_id, name, tag, s3_key, content_hash, docker_image, setup_script, \ + size_bytes, status, build_error, build_job_id, created_by, created_at, updated_at, \ + extra_perms \ + FROM sandbox_snapshot WHERE workspace_id = $1 AND name = $2 AND tag = $3", + &w_id, + &name, + &tag, + ) + .fetch_optional(&db) + .await?; + let row = windmill_common::utils::not_found_if_none(row, "sandbox_snapshot", &name)?; + Ok(Json(row)) +} + +async fn delete_snapshot( + authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, name, tag)): Path<(String, String, String)>, +) -> error::Result { + require_admin(authed.is_admin, &authed.username)?; + + let row = sqlx::query!( + "DELETE FROM sandbox_snapshot WHERE workspace_id = $1 AND name = $2 AND tag = $3 \ + RETURNING s3_key", + &w_id, + &name, + &tag, + ) + .fetch_optional(&db) + .await?; + + if let Some(row) = row { + delete_s3_object(&row.s3_key).await; + } + + Ok(format!("Deleted snapshot {}:{}", name, tag)) +} + +async fn delete_snapshot_all_tags( + authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, name)): Path<(String, String)>, +) -> error::Result { + require_admin(authed.is_admin, &authed.username)?; + + let rows = sqlx::query!( + "DELETE FROM sandbox_snapshot WHERE workspace_id = $1 AND name = $2 RETURNING s3_key", + &w_id, + &name, + ) + .fetch_all(&db) + .await?; + + for row in &rows { + delete_s3_object(&row.s3_key).await; + } + + Ok(format!( + "Deleted {} tag(s) for snapshot {}", + rows.len(), + name + )) +} + +async fn rebuild_snapshot( + authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, name, tag)): Path<(String, String, String)>, +) -> error::Result { + require_admin(authed.is_admin, &authed.username)?; + sqlx::query!( + "UPDATE sandbox_snapshot SET status = 'pending', build_error = NULL, updated_at = now() \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + &w_id, + &name, + &tag, + ) + .execute(&db) + .await?; + Ok(format!("Rebuild queued for snapshot {}:{}", name, tag)) +} + +// --- Volumes --- + +async fn list_volumes( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, +) -> JsonResult> { + require_admin(authed.is_admin, &authed.username)?; + let rows = sqlx::query_as!( + SandboxVolume, + "SELECT workspace_id, name, s3_key, size_bytes, created_by, created_at, updated_at, \ + extra_perms \ + FROM sandbox_volume WHERE workspace_id = $1 ORDER BY created_at DESC", + &w_id, + ) + .fetch_all(&db) + .await?; + Ok(Json(rows)) +} + +#[derive(Deserialize)] +struct CreateVolume { + name: String, +} + +async fn create_volume( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + Json(body): Json, +) -> error::Result { + require_admin(authed.is_admin, &authed.username)?; + let s3_key = format!("sandbox/volumes/{}/{}.tar.gz", w_id, body.name); + sqlx::query!( + "INSERT INTO sandbox_volume (workspace_id, name, s3_key, created_by) \ + VALUES ($1, $2, $3, $4) \ + ON CONFLICT (workspace_id, name) DO NOTHING", + &w_id, + &body.name, + &s3_key, + &authed.username, + ) + .execute(&db) + .await?; + Ok(format!("Volume {} created", body.name)) +} + +async fn get_volume( + authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, name)): Path<(String, String)>, +) -> JsonResult { + require_admin(authed.is_admin, &authed.username)?; + let row = sqlx::query_as!( + SandboxVolume, + "SELECT workspace_id, name, s3_key, size_bytes, created_by, created_at, updated_at, \ + extra_perms \ + FROM sandbox_volume WHERE workspace_id = $1 AND name = $2", + &w_id, + &name, + ) + .fetch_optional(&db) + .await?; + let row = windmill_common::utils::not_found_if_none(row, "sandbox_volume", &name)?; + Ok(Json(row)) +} + +async fn delete_volume( + authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, name)): Path<(String, String)>, +) -> error::Result { + require_admin(authed.is_admin, &authed.username)?; + + let row = sqlx::query!( + "DELETE FROM sandbox_volume WHERE workspace_id = $1 AND name = $2 RETURNING s3_key", + &w_id, + &name, + ) + .fetch_optional(&db) + .await?; + + if let Some(row) = row { + delete_s3_object(&row.s3_key).await; + } + + Ok(format!("Deleted volume {}", name)) +} diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 805bbb2c82..0a8ee2c118 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -84,6 +84,7 @@ pub mod queue; pub mod result_stream; pub mod runnable_settings; pub mod s3_helpers; +pub mod sandbox; pub mod schedule; pub mod schema; pub mod scripts; diff --git a/backend/windmill-common/src/sandbox.rs b/backend/windmill-common/src/sandbox.rs new file mode 100644 index 0000000000..91f309838d --- /dev/null +++ b/backend/windmill-common/src/sandbox.rs @@ -0,0 +1,250 @@ +use std::collections::HashMap; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +#[derive(Debug, Clone, Default)] +pub struct SandboxConfig { + pub snapshot: Option, + pub volumes: HashMap, +} + +#[derive(Debug, Clone)] +pub struct SnapshotRef { + pub name: String, + pub tag: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, sqlx::FromRow)] +pub struct SandboxSnapshot { + pub workspace_id: String, + pub name: String, + pub tag: String, + pub s3_key: String, + pub content_hash: String, + pub docker_image: String, + pub setup_script: Option, + pub size_bytes: Option, + pub status: String, + pub build_error: Option, + pub build_job_id: Option, + pub created_by: String, + pub created_at: DateTime, + pub updated_at: DateTime, + pub extra_perms: serde_json::Value, +} + +#[derive(Serialize, Deserialize, Debug, Clone, sqlx::FromRow)] +pub struct SandboxVolume { + pub workspace_id: String, + pub name: String, + pub s3_key: String, + pub size_bytes: Option, + pub created_by: String, + pub created_at: DateTime, + pub updated_at: DateTime, + pub extra_perms: serde_json::Value, +} + +/// Parse sandbox annotations from script content. +/// +/// Supports comment styles: `#` (Python/bash) and `//` (TS/Go/Rust). +/// +/// ```text +/// # sandbox: python-env:latest +/// # volume: data:/workspace/data +/// # volume: models:/workspace/models +/// +/// // sandbox: node-env:v2 +/// // volume: cache:/tmp/cache +/// ``` +/// +/// Tag defaults to `"latest"` if omitted: `# sandbox: python-env` +pub fn parse_sandbox_config(code: &str) -> SandboxConfig { + let mut config = SandboxConfig::default(); + for line in code.lines() { + let trimmed = line.trim(); + let content = trimmed + .strip_prefix('#') + .or_else(|| trimmed.strip_prefix("//")) + .map(|s| s.trim()); + let Some(content) = content else { + continue; + }; + if let Some(spec) = content.strip_prefix("sandbox:").map(|s| s.trim()) { + let (name, tag) = spec + .split_once(':') + .map(|(n, t)| (n.to_string(), t.to_string())) + .unwrap_or_else(|| (spec.to_string(), "latest".to_string())); + config.snapshot = Some(SnapshotRef { name, tag }); + } else if let Some(spec) = content.strip_prefix("volume:").map(|s| s.trim()) { + if let Some((name, path)) = spec.split_once(':') { + config.volumes.insert(name.to_string(), path.to_string()); + } + } + } + config +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_python_annotations() { + let code = "# sandbox: python-env:latest\n\ + # volume: data:/workspace/data\n\ + # volume: models:/workspace/models\n\ + def main():\n pass\n"; + let config = parse_sandbox_config(code); + let snap = config.snapshot.unwrap(); + assert_eq!(snap.name, "python-env"); + assert_eq!(snap.tag, "latest"); + assert_eq!(config.volumes.len(), 2); + assert_eq!(config.volumes["data"], "/workspace/data"); + assert_eq!(config.volumes["models"], "/workspace/models"); + } + + #[test] + fn test_parse_ts_annotations() { + let code = "// sandbox: node-env:v2\n\ + // volume: cache:/tmp/cache\n\ + export async function main() {}\n"; + let config = parse_sandbox_config(code); + let snap = config.snapshot.unwrap(); + assert_eq!(snap.name, "node-env"); + assert_eq!(snap.tag, "v2"); + assert_eq!(config.volumes["cache"], "/tmp/cache"); + } + + #[test] + fn test_parse_default_tag() { + let code = "# sandbox: myenv\n"; + let config = parse_sandbox_config(code); + let snap = config.snapshot.unwrap(); + assert_eq!(snap.name, "myenv"); + assert_eq!(snap.tag, "latest"); + } + + #[test] + fn test_parse_no_annotations() { + let code = "def main():\n pass\n"; + let config = parse_sandbox_config(code); + assert!(config.snapshot.is_none()); + assert!(config.volumes.is_empty()); + } + + #[test] + fn test_parse_volumes_only() { + let code = "# volume: data:/mnt/data\ndef main(): pass\n"; + let config = parse_sandbox_config(code); + assert!(config.snapshot.is_none()); + assert_eq!(config.volumes.len(), 1); + assert_eq!(config.volumes["data"], "/mnt/data"); + } + + #[test] + fn test_parse_empty_code() { + let config = parse_sandbox_config(""); + assert!(config.snapshot.is_none()); + assert!(config.volumes.is_empty()); + } + + #[test] + fn test_parse_leading_whitespace() { + let code = " # sandbox: indented-env:v1\n # volume: cache:/mnt/cache\n"; + let config = parse_sandbox_config(code); + let snap = config.snapshot.unwrap(); + assert_eq!(snap.name, "indented-env"); + assert_eq!(snap.tag, "v1"); + assert_eq!(config.volumes["cache"], "/mnt/cache"); + } + + #[test] + fn test_parse_mixed_comment_styles() { + let code = "# sandbox: py-env:v3\n// volume: data:/mnt/data\n# volume: logs:/var/log\n"; + let config = parse_sandbox_config(code); + let snap = config.snapshot.unwrap(); + assert_eq!(snap.name, "py-env"); + assert_eq!(snap.tag, "v3"); + assert_eq!(config.volumes.len(), 2); + assert_eq!(config.volumes["data"], "/mnt/data"); + assert_eq!(config.volumes["logs"], "/var/log"); + } + + #[test] + fn test_parse_last_sandbox_wins() { + let code = "# sandbox: first:v1\n# sandbox: second:v2\n"; + let config = parse_sandbox_config(code); + let snap = config.snapshot.unwrap(); + assert_eq!(snap.name, "second"); + assert_eq!(snap.tag, "v2"); + } + + #[test] + fn test_parse_volume_path_with_nested_colons() { + // "volume: name:/some/path" — split_once(':') splits on first colon only + let code = "# volume: db:/opt/db:data\n"; + let config = parse_sandbox_config(code); + // name = "db", path = "/opt/db:data" (everything after first colon) + assert_eq!(config.volumes["db"], "/opt/db:data"); + } + + #[test] + fn test_parse_annotations_among_code() { + let code = "import os\n\ + # sandbox: ml-env:gpu\n\ + # volume: models:/workspace/models\n\ + \n\ + def main():\n\ + # This is a regular comment, not an annotation\n\ + return os.getenv('PATH')\n"; + let config = parse_sandbox_config(code); + let snap = config.snapshot.unwrap(); + assert_eq!(snap.name, "ml-env"); + assert_eq!(snap.tag, "gpu"); + assert_eq!(config.volumes.len(), 1); + assert_eq!(config.volumes["models"], "/workspace/models"); + } + + #[test] + fn test_parse_go_style() { + let code = "// sandbox: go-env:1.22\n// volume: cache:/tmp/go-cache\npackage main\n"; + let config = parse_sandbox_config(code); + let snap = config.snapshot.unwrap(); + assert_eq!(snap.name, "go-env"); + assert_eq!(snap.tag, "1.22"); + assert_eq!(config.volumes["cache"], "/tmp/go-cache"); + } + + #[test] + fn test_parse_bash_style_snapshot_with_multiple_volumes() { + let code = "#!/bin/bash\n\ + # sandbox: custom-runtime\n\ + # volume: input:/mnt/input\n\ + # volume: output:/mnt/output\n\ + # volume: scratch:/tmp/scratch\n\ + echo 'hello'\n"; + let config = parse_sandbox_config(code); + let snap = config.snapshot.unwrap(); + assert_eq!(snap.name, "custom-runtime"); + assert_eq!(snap.tag, "latest"); + assert_eq!(config.volumes.len(), 3); + assert_eq!(config.volumes["input"], "/mnt/input"); + assert_eq!(config.volumes["output"], "/mnt/output"); + assert_eq!(config.volumes["scratch"], "/tmp/scratch"); + } + + #[test] + fn test_parse_ignores_unrelated_comments() { + let code = "# This is a normal comment\n\ + # Another comment\n\ + // Yet another\n\ + # sandboxing is cool (not a directive)\n\ + # volume: data:/mnt/data\n"; + let config = parse_sandbox_config(code); + assert!(config.snapshot.is_none()); + assert_eq!(config.volumes.len(), 1); + } +} diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index d95321dd77..6080803624 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -119,6 +119,8 @@ hex.workspace = true tiberius = { workspace = true, optional = true } tokio-util = { workspace = true, optional = true } tar.workspace = true +flate2.workspace = true +tempfile.workspace = true object_store = { workspace = true, optional = true} convert_case.workspace = true yaml-rust.workspace = true diff --git a/backend/windmill-worker/src/ansible_executor.rs b/backend/windmill-worker/src/ansible_executor.rs index aa250de4ab..01d677d743 100644 --- a/backend/windmill-worker/src/ansible_executor.rs +++ b/backend/windmill-worker/src/ansible_executor.rs @@ -1180,9 +1180,7 @@ mount {{ ) }) .join("\n"); - let _ = write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_ANSIBLE_CONTENT .replace("{PY_INSTALL_DIR}", PY_INSTALL_DIR) .replace("{JOB_DIR}", job_dir) @@ -1194,7 +1192,8 @@ mount {{ "{ADDITIONAL_PYTHON_PATHS}", additional_python_paths_folders.as_str(), ), - )?; + ); + let _ = write_file(job_dir, "run.config.proto", &nsjail_config)?; } else { reserved_variables.insert("PYTHONPATH".to_string(), additional_python_paths_folders); } diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 64270b4044..08d8a29553 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -187,16 +187,15 @@ exit $exit_status // Use nsjail if globally enabled OR if script has #sandbox annotation let nsjail = (is_sandboxing_enabled() || annotation.sandbox) && is_regular_job; let child = if nsjail { - let _ = write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_BASH_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{SHARED_MOUNT}", shared_mount) .replace("{TRACING_PROXY_CA_CERT_PATH}", TRACING_PROXY_CA_CERT_PATH) .replace("#{DEV}", DEV_CONF_NSJAIL), - )?; + ); + let _ = write_file(job_dir, "run.config.proto", &nsjail_config)?; let mut cmd_args = vec![ "--config", "run.config.proto", diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index d8f49456de..47b2b9acd1 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -1451,9 +1451,7 @@ try {{ //do not cache local dependencies let child = if is_sandboxing_enabled() { - let _ = write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_BUN_CONTENT .replace("{LANG}", if annotation.nodejs { "nodejs" } else { "bun" }) .replace("{JOB_DIR}", job_dir) @@ -1471,7 +1469,8 @@ try {{ ) .replace("{TRACING_PROXY_CA_CERT_PATH}", TRACING_PROXY_CA_CERT_PATH) .replace("#{DEV}", DEV_CONF_NSJAIL), - )?; + ); + let _ = write_file(job_dir, "run.config.proto", &nsjail_config)?; let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str()); let args = if annotation.nodejs { diff --git a/backend/windmill-worker/src/csharp_executor.rs b/backend/windmill-worker/src/csharp_executor.rs index 83fdb1aa0e..509ab7d2e4 100644 --- a/backend/windmill-worker/src/csharp_executor.rs +++ b/backend/windmill-worker/src/csharp_executor.rs @@ -584,9 +584,7 @@ pub async fn handle_csharp_job( get_reserved_variables(job, &client.token, conn, parent_runnable_path).await?; let child = if is_sandboxing_enabled() { - write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_CSHARP_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{CACHE_DIR}", CSHARP_CACHE_DIR) @@ -595,7 +593,8 @@ pub async fn handle_csharp_job( .replace("{SHARED_MOUNT}", shared_mount) .replace("{TRACING_PROXY_CA_CERT_PATH}", TRACING_PROXY_CA_CERT_PATH) .replace("#{DEV}", DEV_CONF_NSJAIL), - )?; + ); + write_file(job_dir, "run.config.proto", &nsjail_config)?; let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str()); nsjail_cmd .current_dir(job_dir) diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 60e91346e2..2c81d1a678 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -339,16 +339,15 @@ func Run(req Req) (interface{{}}, error){{ get_reserved_variables(job, &client.token, conn, parent_runnable_path).await?; let child = if is_sandboxing_enabled() { - let _ = write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_GO_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{SHARED_MOUNT}", shared_mount) .replace("{TRACING_PROXY_CA_CERT_PATH}", TRACING_PROXY_CA_CERT_PATH) .replace("#{DEV}", DEV_CONF_NSJAIL), - )?; + ); + let _ = write_file(job_dir, "run.config.proto", &nsjail_config)?; let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str()); nsjail_cmd .current_dir(job_dir) diff --git a/backend/windmill-worker/src/java_executor.rs b/backend/windmill-worker/src/java_executor.rs index c12957fffc..aae9247bf2 100644 --- a/backend/windmill-worker/src/java_executor.rs +++ b/backend/windmill-worker/src/java_executor.rs @@ -598,16 +598,14 @@ async fn run<'a>( ) .await; - write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_JAVA_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{CACHE_DIR}", JAVA_CACHE_DIR) .replace("{SHARED_MOUNT}", &shared_mount) - // .replace("{CACHED_TARGET}", &shared_mount) .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()), - )?; + ); + write_file(job_dir, "run.config.proto", &nsjail_config)?; let mut cmd = Command::new(NSJAIL_PATH.as_str()); cmd.env_clear() .current_dir(job_dir) diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 741458c0e0..1b66cb9d31 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -66,6 +66,7 @@ mod python_versions; pub mod result_processor; #[cfg(feature = "rust")] mod rust_executor; +mod sandbox_setup; mod sanitized_sql_params; mod schema; pub mod sql_utils; diff --git a/backend/windmill-worker/src/nu_executor.rs b/backend/windmill-worker/src/nu_executor.rs index a4320a3612..a46d03ba5e 100644 --- a/backend/windmill-worker/src/nu_executor.rs +++ b/backend/windmill-worker/src/nu_executor.rs @@ -245,9 +245,7 @@ async fn run<'a>( ) .await; - write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_NU_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{NU_PATH}", &NU_PATH) @@ -255,7 +253,8 @@ async fn run<'a>( .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{TRACING_PROXY_CA_CERT_PATH}", TRACING_PROXY_CA_CERT_PATH) .replace("#{DEV}", DEV_CONF_NSJAIL), - )?; + ); + write_file(job_dir, "run.config.proto", &nsjail_config)?; let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str()); nsjail_cmd .env_clear() diff --git a/backend/windmill-worker/src/php_executor.rs b/backend/windmill-worker/src/php_executor.rs index e98755e555..771f999745 100644 --- a/backend/windmill-worker/src/php_executor.rs +++ b/backend/windmill-worker/src/php_executor.rs @@ -294,14 +294,13 @@ try {{ let (reserved_variables, _) = tokio::try_join!(reserved_variables_args_out_f, write_wrapper_f)?; let child = if is_sandboxing_enabled() { - let _ = write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_PHP_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{SHARED_MOUNT}", shared_mount), - )?; + ); + let _ = write_file(job_dir, "run.config.proto", &nsjail_config)?; let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str()); let args = vec![ diff --git a/backend/windmill-worker/src/pwsh_executor.rs b/backend/windmill-worker/src/pwsh_executor.rs index acae81a588..41f5cd4c43 100644 --- a/backend/windmill-worker/src/pwsh_executor.rs +++ b/backend/windmill-worker/src/pwsh_executor.rs @@ -518,15 +518,14 @@ $env:PSModulePath = \"{};$PSModulePathBackup\"", let nsjail = is_sandboxing_enabled() && is_regular_job; let child = if nsjail { - let _ = write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_POWERSHELL_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{SHARED_MOUNT}", shared_mount) .replace("{CACHE_DIR}", POWERSHELL_CACHE_DIR), - )?; + ); + let _ = write_file(job_dir, "run.config.proto", &nsjail_config)?; let cmd_args = vec![ "--config", "run.config.proto", diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index a701896db1..2ded2c1323 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -800,9 +800,7 @@ mount {{ ) }) .join("\n"); - let _ = write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_PYTHON3_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{PY_INSTALL_DIR}", PY_INSTALL_DIR) @@ -817,7 +815,8 @@ mount {{ ) .replace("{TRACING_PROXY_CA_CERT_PATH}", TRACING_PROXY_CA_CERT_PATH) .replace("#{DEV}", DEV_CONF_NSJAIL), - )?; + ); + let _ = write_file(job_dir, "run.config.proto", &nsjail_config)?; } else { reserved_variables.insert("PYTHONPATH".to_string(), additional_python_paths_folders); } diff --git a/backend/windmill-worker/src/ruby_executor.rs b/backend/windmill-worker/src/ruby_executor.rs index 869b58d7de..4346f60e6b 100644 --- a/backend/windmill-worker/src/ruby_executor.rs +++ b/backend/windmill-worker/src/ruby_executor.rs @@ -793,9 +793,7 @@ mount {{ }) .join("\n"); - write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_RUBY_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{SHARED_MOUNT}", &shared_mount) @@ -803,7 +801,8 @@ mount {{ .replace("{TRACING_PROXY_CA_CERT_PATH}", TRACING_PROXY_CA_CERT_PATH) .replace("#{DEV}", DEV_CONF_NSJAIL) .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()), - )?; + ); + write_file(job_dir, "run.config.proto", &nsjail_config)?; let mut cmd = Command::new(NSJAIL_PATH.as_str()); cmd.env_clear() .current_dir(job_dir) diff --git a/backend/windmill-worker/src/rust_executor.rs b/backend/windmill-worker/src/rust_executor.rs index 87db45fb64..12d78669c4 100644 --- a/backend/windmill-worker/src/rust_executor.rs +++ b/backend/windmill-worker/src/rust_executor.rs @@ -663,9 +663,7 @@ pub async fn handle_rust_job( append_logs(&job.id, &job.workspace_id, logs2, conn).await; let child = if is_sandboxing_enabled() { - let _ = write_file( - job_dir, - "run.config.proto", + let nsjail_config = crate::sandbox_setup::finalize_nsjail_config( &NSJAIL_CONFIG_RUN_RUST_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{CACHE_DIR}", RUST_CACHE_DIR) @@ -674,7 +672,8 @@ pub async fn handle_rust_job( .replace("{TRACING_PROXY_CA_CERT_PATH}", TRACING_PROXY_CA_CERT_PATH) .replace("#{DEV}", DEV_CONF_NSJAIL) .replace("{SHARED_MOUNT}", shared_mount), - )?; + ); + let _ = write_file(job_dir, "run.config.proto", &nsjail_config)?; let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str()); nsjail_cmd .current_dir(job_dir) diff --git a/backend/windmill-worker/src/sandbox_setup.rs b/backend/windmill-worker/src/sandbox_setup.rs new file mode 100644 index 0000000000..81a0d8b574 --- /dev/null +++ b/backend/windmill-worker/src/sandbox_setup.rs @@ -0,0 +1,1009 @@ +use std::{ + collections::HashMap, + path::{Path, PathBuf}, +}; + +use windmill_common::error::{self, Error}; + +const OVERLAY_MARKER: &str = "# SANDBOX_OVERLAY_ACTIVE"; + +#[derive(Default)] +pub struct SandboxSetupState { + pub overlay: Option, + /// name → (local_dir, mount_path) + pub volume_mounts: HashMap, +} + +pub struct OverlayMount { + pub merged: PathBuf, + pub upper: PathBuf, + pub work: PathBuf, +} + +/// Build additional nsjail mount blocks for sandbox config. +/// Returns a string to be appended to the `shared_mount` variable. +/// If an overlay is present, includes a marker comment that `finalize_nsjail_config` +/// uses to strip system directory mounts from the final config. +pub fn build_sandbox_mounts(setup: &SandboxSetupState) -> String { + let mut mounts = String::new(); + + if let Some(ref overlay) = setup.overlay { + mounts.push_str(&format!( + "\n{OVERLAY_MARKER}\nmount {{\n src: \"{}\"\n dst: \"/\"\n is_bind: true\n rw: true\n}}\n", + overlay.merged.display() + )); + } + + for (_name, (local_dir, mount_path)) in &setup.volume_mounts { + mounts.push_str(&format!( + "\nmount {{\n src: \"{}\"\n dst: \"{}\"\n is_bind: true\n rw: true\n}}\n", + local_dir.display(), + mount_path + )); + } + + mounts +} + +/// Post-process an nsjail config string. If the overlay marker is present, +/// remove bind mounts for system directories (/bin, /lib, /lib64, /usr, /etc) +/// since the overlay root mount provides them. +/// If no overlay marker is present, returns the config unchanged. +pub fn finalize_nsjail_config(config: &str) -> String { + if !config.contains(OVERLAY_MARKER) { + return config.to_string(); + } + + let system_dirs: &[&str] = &["/bin", "/lib", "/lib64", "/usr", "/etc"]; + let mut result_lines: Vec<&str> = Vec::new(); + let mut in_mount_block = false; + let mut mount_block_start = 0; + let mut should_remove_block = false; + + let lines: Vec<&str> = config.lines().collect(); + let mut i = 0; + + while i < lines.len() { + let trimmed = lines[i].trim(); + + if trimmed == OVERLAY_MARKER { + i += 1; + continue; + } + + if !in_mount_block && (trimmed.starts_with("mount {") || trimmed == "mount {") { + in_mount_block = true; + mount_block_start = result_lines.len(); + should_remove_block = false; + result_lines.push(lines[i]); + i += 1; + continue; + } + + if in_mount_block { + if trimmed.starts_with("dst: \"") { + if let Some(dst) = trimmed + .strip_prefix("dst: \"") + .and_then(|s| s.strip_suffix('"')) + { + if system_dirs.contains(&dst) { + should_remove_block = true; + } + } + } + + if trimmed == "}" { + if should_remove_block { + result_lines.truncate(mount_block_start); + } else { + result_lines.push(lines[i]); + } + in_mount_block = false; + i += 1; + continue; + } + + if !should_remove_block { + result_lines.push(lines[i]); + } + i += 1; + continue; + } + + result_lines.push(lines[i]); + i += 1; + } + + result_lines.join("\n") +} + +#[cfg(feature = "parquet")] +const SNAPSHOT_CACHE_DIR: &str = "/tmp/windmill/snapshots"; + +/// Download snapshot from S3 if not cached locally. Returns path to unpacked rootfs. +#[cfg(feature = "parquet")] +pub async fn ensure_snapshot_cached( + w_id: &str, + name: &str, + tag: &str, + db: &windmill_common::DB, +) -> error::Result { + use windmill_common::s3_helpers::{attempt_fetch_bytes, get_object_store}; + + let row = sqlx::query!( + "SELECT s3_key, content_hash, status FROM sandbox_snapshot \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + w_id, + name, + tag, + ) + .fetch_optional(db) + .await?; + + let row = row.ok_or_else(|| { + Error::NotFound(format!( + "sandbox snapshot {name}:{tag} not found in workspace {w_id}" + )) + })?; + + if row.status != "ready" { + return Err(Error::ExecutionErr(format!( + "sandbox snapshot {name}:{tag} is not ready (status: {})", + row.status + ))); + } + + let cache_key = if row.content_hash.is_empty() { + format!("{w_id}_{name}_{tag}") + } else { + row.content_hash.clone() + }; + let cache_path = PathBuf::from(SNAPSHOT_CACHE_DIR).join(&cache_key); + + if cache_path.exists() { + tracing::info!("Snapshot {name}:{tag} found in cache at {}", cache_path.display()); + return Ok(cache_path); + } + + let os = get_object_store() + .await + .ok_or_else(|| Error::ExecutionErr("S3 object store not configured".to_string()))?; + + tracing::info!( + "Downloading snapshot {name}:{tag} from S3 key: {}", + row.s3_key + ); + let bytes = attempt_fetch_bytes(os, &row.s3_key).await?; + + tokio::fs::create_dir_all(&cache_path) + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to create snapshot cache dir: {e}")))?; + + let cache_path_clone = cache_path.clone(); + let bytes_clone = bytes.to_vec(); + tokio::task::spawn_blocking(move || -> error::Result<()> { + use flate2::read::GzDecoder; + use std::io::Cursor; + use tar::Archive; + + let decoder = GzDecoder::new(Cursor::new(bytes_clone)); + let mut archive = Archive::new(decoder); + archive + .unpack(&cache_path_clone) + .map_err(|e| Error::ExecutionErr(format!("Failed to unpack snapshot tar.gz: {e}")))?; + Ok(()) + }) + .await + .map_err(|e| Error::ExecutionErr(format!("Spawn blocking failed: {e}")))??; + + tracing::info!( + "Snapshot {name}:{tag} unpacked to {}", + cache_path.display() + ); + Ok(cache_path) +} + +#[cfg(not(feature = "parquet"))] +pub async fn ensure_snapshot_cached( + _w_id: &str, + _name: &str, + _tag: &str, + _db: &windmill_common::DB, +) -> error::Result { + Err(Error::ExecutionErr( + "Sandbox snapshots require the parquet feature (S3 object store)".to_string(), + )) +} + +/// Mount overlayfs: lower=snapshot (read-only), upper+work=per-job writable. +pub async fn mount_overlay( + snapshot_path: &Path, + job_dir: &str, +) -> error::Result { + use tokio::process::Command; + + let upper = PathBuf::from(job_dir).join("overlay_upper"); + let work = PathBuf::from(job_dir).join("overlay_work"); + let merged = PathBuf::from(job_dir).join("overlay_merged"); + + for dir in [&upper, &work, &merged] { + tokio::fs::create_dir_all(dir) + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to create overlay dir: {e}")))?; + } + + let mount_opts = format!( + "lowerdir={},upperdir={},workdir={}", + snapshot_path.display(), + upper.display(), + work.display() + ); + + let output = Command::new("mount") + .args([ + "-t", + "overlay", + "overlay", + "-o", + &mount_opts, + &merged.to_string_lossy(), + ]) + .output() + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to run mount: {e}")))?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(Error::ExecutionErr(format!( + "Failed to mount overlayfs: {stderr}" + ))); + } + + tracing::info!("Overlayfs mounted at {}", merged.display()); + Ok(OverlayMount { + merged, + upper, + work, + }) +} + +/// Unmount overlayfs and clean up per-job dirs. +pub async fn unmount_overlay(overlay: &OverlayMount) -> error::Result<()> { + use tokio::process::Command; + + let output = Command::new("umount") + .arg(&overlay.merged) + .output() + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to run umount: {e}")))?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + tracing::warn!("Failed to umount overlayfs at {}: {stderr}", overlay.merged.display()); + } + + for dir in [&overlay.merged, &overlay.upper, &overlay.work] { + if let Err(e) = tokio::fs::remove_dir_all(dir).await { + tracing::warn!("Failed to clean up overlay dir {}: {e}", dir.display()); + } + } + + Ok(()) +} + +/// Download volume from S3 to local dir. If no S3 object exists, create empty dir (auto-create). +#[cfg(feature = "parquet")] +pub async fn download_volume( + w_id: &str, + name: &str, + local_path: &Path, + db: &windmill_common::DB, +) -> error::Result<()> { + use windmill_common::s3_helpers::{attempt_fetch_bytes, get_object_store}; + + let row = sqlx::query!( + "SELECT s3_key FROM sandbox_volume WHERE workspace_id = $1 AND name = $2", + w_id, + name, + ) + .fetch_optional(db) + .await?; + + let s3_key = match row { + Some(r) => r.s3_key, + None => { + let s3_key = format!("sandbox/volumes/{w_id}/{name}.tar.gz"); + sqlx::query!( + "INSERT INTO sandbox_volume (workspace_id, name, s3_key, created_by) \ + VALUES ($1, $2, $3, 'system') ON CONFLICT DO NOTHING", + w_id, + name, + &s3_key, + ) + .execute(db) + .await?; + tracing::info!("Auto-created volume {name} in workspace {w_id}"); + return Ok(()); + } + }; + + let Some(os) = get_object_store().await else { + tracing::info!("No object store configured, using empty volume for {name}"); + return Ok(()); + }; + + let bytes = match attempt_fetch_bytes(os, &s3_key).await { + Ok(b) => b, + Err(_) => { + tracing::info!("Volume {name} has no S3 data yet, using empty dir"); + return Ok(()); + } + }; + + let local_path = local_path.to_path_buf(); + let bytes_vec = bytes.to_vec(); + tokio::task::spawn_blocking(move || -> error::Result<()> { + use flate2::read::GzDecoder; + use std::io::Cursor; + use tar::Archive; + + let decoder = GzDecoder::new(Cursor::new(bytes_vec)); + let mut archive = Archive::new(decoder); + archive + .unpack(&local_path) + .map_err(|e| Error::ExecutionErr(format!("Failed to unpack volume tar.gz: {e}")))?; + Ok(()) + }) + .await + .map_err(|e| Error::ExecutionErr(format!("Spawn blocking failed: {e}")))??; + + tracing::info!("Volume {name} downloaded and unpacked"); + Ok(()) +} + +#[cfg(not(feature = "parquet"))] +pub async fn download_volume( + _w_id: &str, + _name: &str, + _local_path: &Path, + _db: &windmill_common::DB, +) -> error::Result<()> { + Err(Error::ExecutionErr( + "Sandbox volumes require the parquet feature (S3 object store)".to_string(), + )) +} + +/// Re-tar volume dir, upload to S3. Update size_bytes/updated_at in DB. +#[cfg(feature = "parquet")] +pub async fn upload_volume( + w_id: &str, + name: &str, + local_path: &Path, + db: &windmill_common::DB, +) -> error::Result<()> { + use windmill_common::s3_helpers::get_object_store; + + let Some(os) = get_object_store().await else { + tracing::warn!("No object store configured, cannot upload volume {name}"); + return Ok(()); + }; + + let s3_key = sqlx::query_scalar!( + "SELECT s3_key FROM sandbox_volume WHERE workspace_id = $1 AND name = $2", + w_id, + name, + ) + .fetch_optional(db) + .await? + .ok_or_else(|| Error::NotFound(format!("sandbox volume {name} not found")))?; + + let local_path = local_path.to_path_buf(); + let bytes = tokio::task::spawn_blocking(move || -> error::Result> { + use flate2::write::GzEncoder; + use flate2::Compression; + use tar::Builder; + + let buf = Vec::new(); + let encoder = GzEncoder::new(buf, Compression::fast()); + let mut builder = Builder::new(encoder); + builder + .append_dir_all(".", &local_path) + .map_err(|e| Error::ExecutionErr(format!("Failed to tar volume: {e}")))?; + let encoder = builder + .into_inner() + .map_err(|e| Error::ExecutionErr(format!("Failed to finish tar: {e}")))?; + let bytes = encoder + .finish() + .map_err(|e| Error::ExecutionErr(format!("Failed to finish gzip: {e}")))?; + Ok(bytes) + }) + .await + .map_err(|e| Error::ExecutionErr(format!("Spawn blocking failed: {e}")))??; + + let size = bytes.len() as i64; + + os.put( + &object_store::path::Path::from(s3_key.as_str()), + bytes.into(), + ) + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to upload volume to S3: {e}")))?; + + sqlx::query!( + "UPDATE sandbox_volume SET size_bytes = $3, updated_at = now() \ + WHERE workspace_id = $1 AND name = $2", + w_id, + name, + size, + ) + .execute(db) + .await?; + + tracing::info!("Volume {name} uploaded to S3 ({size} bytes)"); + Ok(()) +} + +#[cfg(not(feature = "parquet"))] +pub async fn upload_volume( + _w_id: &str, + _name: &str, + _local_path: &Path, + _db: &windmill_common::DB, +) -> error::Result<()> { + Err(Error::ExecutionErr( + "Sandbox volumes require the parquet feature (S3 object store)".to_string(), + )) +} + +/// Pull Docker image + run optional setup script → tar.gz → upload to S3. +#[cfg(feature = "parquet")] +#[allow(dead_code)] +pub async fn build_snapshot( + w_id: &str, + name: &str, + tag: &str, + docker_image: &str, + setup_script: Option<&str>, + db: &windmill_common::DB, +) -> error::Result<()> { + use tokio::process::Command; + use windmill_common::s3_helpers::get_object_store; + + sqlx::query!( + "UPDATE sandbox_snapshot SET status = 'building', updated_at = now() \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + w_id, + name, + tag, + ) + .execute(db) + .await?; + + let result = async { + let temp_dir = tempfile::tempdir() + .map_err(|e| Error::ExecutionErr(format!("Failed to create temp dir: {e}")))?; + let rootfs_dir = temp_dir.path().join("rootfs"); + tokio::fs::create_dir_all(&rootfs_dir) + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to create rootfs dir: {e}")))?; + + tracing::info!("Exporting docker image {docker_image} for snapshot {name}:{tag}"); + let crane_output = Command::new("crane") + .args(["export", docker_image, "-"]) + .output() + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to run crane: {e}")))?; + + if !crane_output.status.success() { + let stderr = String::from_utf8_lossy(&crane_output.stderr); + return Err(Error::ExecutionErr(format!("crane export failed: {stderr}"))); + } + + let crane_bytes = crane_output.stdout; + let rootfs_dir_clone = rootfs_dir.clone(); + tokio::task::spawn_blocking(move || -> error::Result<()> { + use std::io::Cursor; + use tar::Archive; + let mut archive = Archive::new(Cursor::new(crane_bytes)); + archive + .unpack(&rootfs_dir_clone) + .map_err(|e| Error::ExecutionErr(format!("Failed to unpack crane output: {e}")))?; + Ok(()) + }) + .await + .map_err(|e| Error::ExecutionErr(format!("Spawn blocking failed: {e}")))??; + + if let Some(script) = setup_script { + if !script.trim().is_empty() { + tracing::info!("Running setup script for snapshot {name}:{tag}"); + let output = Command::new("chroot") + .args([ + &rootfs_dir.to_string_lossy().to_string(), + "/bin/sh", + "-c", + script, + ]) + .output() + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to run setup: {e}")))?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(Error::ExecutionErr(format!("Setup script failed: {stderr}"))); + } + } + } + + let rootfs_dir_clone = rootfs_dir.clone(); + let (bytes, content_hash) = + tokio::task::spawn_blocking(move || -> error::Result<(Vec, String)> { + use flate2::write::GzEncoder; + use flate2::Compression; + use sha2::{Digest, Sha256}; + use tar::Builder; + + let buf = Vec::new(); + let encoder = GzEncoder::new(buf, Compression::default()); + let mut builder = Builder::new(encoder); + builder + .append_dir_all(".", &rootfs_dir_clone) + .map_err(|e| Error::ExecutionErr(format!("Failed to tar snapshot: {e}")))?; + let encoder = builder + .into_inner() + .map_err(|e| Error::ExecutionErr(format!("Failed to finish tar: {e}")))?; + let bytes = encoder + .finish() + .map_err(|e| Error::ExecutionErr(format!("Failed to finish gzip: {e}")))?; + let hash = format!("{:x}", Sha256::digest(&bytes)); + Ok((bytes, hash)) + }) + .await + .map_err(|e| Error::ExecutionErr(format!("Spawn blocking failed: {e}")))??; + + let s3_key = format!("sandbox/snapshots/{w_id}/{name}/{content_hash}.tar.gz"); + let size = bytes.len() as i64; + + let os = get_object_store() + .await + .ok_or_else(|| Error::ExecutionErr("S3 object store not configured".to_string()))?; + + os.put( + &object_store::path::Path::from(s3_key.as_str()), + bytes.into(), + ) + .await + .map_err(|e| Error::ExecutionErr(format!("Failed to upload snapshot to S3: {e}")))?; + + sqlx::query!( + "UPDATE sandbox_snapshot SET \ + s3_key = $4, content_hash = $5, size_bytes = $6, status = 'ready', \ + build_error = NULL, updated_at = now() \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + w_id, + name, + tag, + &s3_key, + &content_hash, + size, + ) + .execute(db) + .await?; + + tracing::info!("Snapshot {name}:{tag} built successfully ({size} bytes)"); + Ok(()) + } + .await; + + if let Err(ref e) = result { + let err_str = e.to_string(); + sqlx::query!( + "UPDATE sandbox_snapshot SET status = 'failed', build_error = $4, updated_at = now() \ + WHERE workspace_id = $1 AND name = $2 AND tag = $3", + w_id, + name, + tag, + &err_str, + ) + .execute(db) + .await + .ok(); + } + + result +} + +#[cfg(test)] +mod tests { + use super::*; + use windmill_common::sandbox::parse_sandbox_config; + + // --- Unit tests for build_sandbox_mounts --- + + #[test] + fn test_build_sandbox_mounts_empty() { + let setup = SandboxSetupState::default(); + let mounts = build_sandbox_mounts(&setup); + assert!(mounts.is_empty()); + } + + #[test] + fn test_build_sandbox_mounts_volumes_only() { + let mut setup = SandboxSetupState::default(); + setup.volume_mounts.insert( + "data".to_string(), + (PathBuf::from("/job/volumes/data"), "/workspace/data".to_string()), + ); + let mounts = build_sandbox_mounts(&setup); + assert!(mounts.contains("dst: \"/workspace/data\"")); + assert!(mounts.contains("src: \"/job/volumes/data\"")); + assert!(mounts.contains("is_bind: true")); + assert!(mounts.contains("rw: true")); + assert!(!mounts.contains(OVERLAY_MARKER)); + } + + #[test] + fn test_build_sandbox_mounts_multiple_volumes() { + let mut setup = SandboxSetupState::default(); + setup.volume_mounts.insert( + "input".to_string(), + (PathBuf::from("/job/volumes/input"), "/mnt/input".to_string()), + ); + setup.volume_mounts.insert( + "output".to_string(), + (PathBuf::from("/job/volumes/output"), "/mnt/output".to_string()), + ); + let mounts = build_sandbox_mounts(&setup); + assert!(mounts.contains("/mnt/input")); + assert!(mounts.contains("/mnt/output")); + assert_eq!(mounts.matches("mount {").count(), 2); + } + + #[test] + fn test_build_sandbox_mounts_overlay_only() { + let setup = SandboxSetupState { + overlay: Some(OverlayMount { + merged: PathBuf::from("/job/overlay_merged"), + upper: PathBuf::from("/job/overlay_upper"), + work: PathBuf::from("/job/overlay_work"), + }), + volume_mounts: HashMap::new(), + }; + let mounts = build_sandbox_mounts(&setup); + assert!(mounts.contains(OVERLAY_MARKER)); + assert!(mounts.contains("src: \"/job/overlay_merged\"")); + assert!(mounts.contains("dst: \"/\"")); + assert!(mounts.contains("is_bind: true")); + assert!(mounts.contains("rw: true")); + } + + #[test] + fn test_build_sandbox_mounts_overlay_and_volumes() { + let mut setup = SandboxSetupState { + overlay: Some(OverlayMount { + merged: PathBuf::from("/job/overlay_merged"), + upper: PathBuf::from("/job/overlay_upper"), + work: PathBuf::from("/job/overlay_work"), + }), + volume_mounts: HashMap::new(), + }; + setup.volume_mounts.insert( + "data".to_string(), + (PathBuf::from("/job/volumes/data"), "/workspace/data".to_string()), + ); + let mounts = build_sandbox_mounts(&setup); + assert!(mounts.contains(OVERLAY_MARKER)); + assert!(mounts.contains("dst: \"/\"")); + assert!(mounts.contains("dst: \"/workspace/data\"")); + assert_eq!(mounts.matches("mount {").count(), 2); + } + + // --- Unit tests for finalize_nsjail_config --- + + #[test] + fn test_finalize_no_overlay_passes_through() { + let config = "name: \"test\"\n\ + mount {\n src: \"/bin\"\n dst: \"/bin\"\n is_bind: true\n}\n\ + mount {\n dst: \"/tmp\"\n fstype: \"tmpfs\"\n rw: true\n}\n"; + let result = finalize_nsjail_config(config); + assert_eq!(result, config); + } + + #[test] + fn test_finalize_strips_all_system_dirs() { + let config = "name: \"test\"\n\ + mount {\n src: \"/bin\"\n dst: \"/bin\"\n is_bind: true\n}\n\ + mount {\n src: \"/lib\"\n dst: \"/lib\"\n is_bind: true\n}\n\ + mount {\n src: \"/lib64\"\n dst: \"/lib64\"\n is_bind: true\n mandatory: false\n}\n\ + mount {\n src: \"/usr\"\n dst: \"/usr\"\n is_bind: true\n}\n\ + mount {\n src: \"/etc\"\n dst: \"/etc\"\n is_bind: true\n}\n\ + # SANDBOX_OVERLAY_ACTIVE\n\ + mount {\n src: \"/job/merged\"\n dst: \"/\"\n is_bind: true\n rw: true\n}\n\ + mount {\n dst: \"/tmp\"\n fstype: \"tmpfs\"\n rw: true\n}\n"; + let result = finalize_nsjail_config(config); + for dir in &["/bin", "/lib", "/lib64", "/usr", "/etc"] { + assert!( + !result.contains(&format!("dst: \"{dir}\"")), + "System dir {dir} should be stripped" + ); + } + assert!(result.contains("dst: \"/\""), "Overlay root mount preserved"); + assert!(result.contains("dst: \"/tmp\""), "tmpfs mount preserved"); + } + + #[test] + fn test_finalize_preserves_non_system_mounts() { + let config = "name: \"test\"\n\ + mount {\n src: \"/bin\"\n dst: \"/bin\"\n is_bind: true\n}\n\ + mount {\n src: \"/dev/null\"\n dst: \"/dev/null\"\n is_bind: true\n rw: true\n}\n\ + mount {\n src: \"/dev/random\"\n dst: \"/dev/random\"\n is_bind: true\n}\n\ + mount {\n src: \"/dev/urandom\"\n dst: \"/dev/urandom\"\n is_bind: true\n}\n\ + mount {\n src: \"/opt/microsoft\"\n dst: \"/opt/microsoft\"\n is_bind: true\n}\n\ + mount {\n src: \"/job/result.json\"\n dst: \"/tmp/result.json\"\n is_bind: true\n rw: true\n}\n\ + mount {\n dst: \"/tmp\"\n fstype: \"tmpfs\"\n rw: true\n}\n\ + mount {\n dst: \"/dev/shm\"\n fstype: \"tmpfs\"\n rw: true\n}\n\ + # SANDBOX_OVERLAY_ACTIVE\n\ + mount {\n src: \"/job/merged\"\n dst: \"/\"\n is_bind: true\n rw: true\n}\n"; + let result = finalize_nsjail_config(config); + assert!(!result.contains("dst: \"/bin\""), "/bin stripped"); + assert!(result.contains("dst: \"/dev/null\""), "/dev/null kept"); + assert!(result.contains("dst: \"/dev/random\""), "/dev/random kept"); + assert!(result.contains("dst: \"/dev/urandom\""), "/dev/urandom kept"); + assert!(result.contains("dst: \"/opt/microsoft\""), "/opt/microsoft kept"); + assert!(result.contains("dst: \"/tmp/result.json\""), "result.json kept"); + assert!(result.contains("dst: \"/tmp\""), "/tmp tmpfs kept"); + assert!(result.contains("dst: \"/dev/shm\""), "/dev/shm kept"); + } + + #[test] + fn test_finalize_removes_overlay_marker_line() { + let config = "name: \"test\"\n\ + # SANDBOX_OVERLAY_ACTIVE\n\ + mount {\n src: \"/job/merged\"\n dst: \"/\"\n is_bind: true\n}\n"; + let result = finalize_nsjail_config(config); + assert!(!result.contains(OVERLAY_MARKER)); + assert!(result.contains("dst: \"/\"")); + } + + // --- End-to-end: real bash proto template with sandbox --- + + #[test] + fn test_e2e_bash_with_snapshot_and_volume() { + // Simulate what happens in bash_executor.rs: + // 1. Parse sandbox annotations from script code + let bash_script = "#!/bin/bash\n\ + # sandbox: python-ml:gpu\n\ + # volume: data:/workspace/data\n\ + # volume: models:/workspace/models\n\ + echo 'running ML pipeline'\n"; + + let sandbox_config = parse_sandbox_config(bash_script); + let snap = sandbox_config.snapshot.as_ref().unwrap(); + assert_eq!(snap.name, "python-ml"); + assert_eq!(snap.tag, "gpu"); + assert_eq!(sandbox_config.volumes.len(), 2); + + // 2. Build sandbox mounts (simulate worker.rs setup) + let setup = SandboxSetupState { + overlay: Some(OverlayMount { + merged: PathBuf::from("/tmp/job123/overlay_merged"), + upper: PathBuf::from("/tmp/job123/overlay_upper"), + work: PathBuf::from("/tmp/job123/overlay_work"), + }), + volume_mounts: HashMap::from([ + ( + "data".to_string(), + (PathBuf::from("/tmp/job123/volumes/data"), "/workspace/data".to_string()), + ), + ( + "models".to_string(), + (PathBuf::from("/tmp/job123/volumes/models"), "/workspace/models".to_string()), + ), + ]), + }; + + let sandbox_mounts = build_sandbox_mounts(&setup); + + // 3. Simulate executor: template replacement + append sandbox mounts + let raw_config = include_str!("../nsjail/run.bash.config.proto"); + let config = raw_config + .replace("{JOB_DIR}", "/tmp/job123") + .replace("{CLONE_NEWUSER}", "true") + .replace("{SHARED_MOUNT}", &sandbox_mounts) + .replace("{TRACING_PROXY_CA_CERT_PATH}", "/dev/null"); + + // 4. Finalize: strip system dir mounts since overlay provides rootfs + let final_config = finalize_nsjail_config(&config); + + // Verify system dirs removed + for dir in &["/bin", "/lib", "/lib64", "/usr", "/etc"] { + assert!( + !final_config.contains(&format!("dst: \"{dir}\"")), + "System dir {dir} should be removed when overlay is active" + ); + } + + // Verify overlay root mount present + assert!(final_config.contains("dst: \"/\"")); + assert!(final_config.contains("src: \"/tmp/job123/overlay_merged\"")); + + // Verify volume mounts present + assert!(final_config.contains("dst: \"/workspace/data\"")); + assert!(final_config.contains("src: \"/tmp/job123/volumes/data\"")); + assert!(final_config.contains("dst: \"/workspace/models\"")); + assert!(final_config.contains("src: \"/tmp/job123/volumes/models\"")); + + // Verify essential non-system mounts preserved + assert!(final_config.contains("dst: \"/tmp\"")); + assert!(final_config.contains("dst: \"/dev/null\"")); + assert!(final_config.contains("dst: \"/dev/random\"")); + assert!(final_config.contains("dst: \"/dev/urandom\"")); + assert!(final_config.contains("dst: \"/tmp/main.sh\"")); + assert!(final_config.contains("dst: \"/tmp/wrapper.sh\"")); + assert!(final_config.contains("dst: \"/tmp/result.json\"")); + + // Verify nsjail metadata preserved + assert!(final_config.contains("name: \"bash run script\"")); + assert!(final_config.contains("mode: ONCE")); + assert!(final_config.contains("envar: \"HOME=/tmp\"")); + } + + // --- End-to-end: real python proto template with sandbox --- + + #[test] + fn test_e2e_python_with_snapshot_only() { + let python_script = "# sandbox: python312:latest\n\ + import pandas as pd\n\ + def main():\n return pd.DataFrame()\n"; + + let sandbox_config = parse_sandbox_config(python_script); + assert!(sandbox_config.snapshot.is_some()); + assert!(sandbox_config.volumes.is_empty()); + + let setup = SandboxSetupState { + overlay: Some(OverlayMount { + merged: PathBuf::from("/tmp/jobXYZ/overlay_merged"), + upper: PathBuf::from("/tmp/jobXYZ/overlay_upper"), + work: PathBuf::from("/tmp/jobXYZ/overlay_work"), + }), + volume_mounts: HashMap::new(), + }; + + let sandbox_mounts = build_sandbox_mounts(&setup); + let raw_config = include_str!("../nsjail/run.python3.config.proto"); + let config = raw_config + .replace("{JOB_DIR}", "/tmp/jobXYZ") + .replace("{MAIN}", "main") + .replace("{CLONE_NEWUSER}", "true") + .replace("{SHARED_MOUNT}", &sandbox_mounts) + .replace("{SHARED_DEPENDENCIES}", "") + .replace("{PY_INSTALL_DIR}", "/usr/local") + .replace("{GLOBAL_SITE_PACKAGES}", "/usr/lib/python3/dist-packages") + .replace("{ADDITIONAL_PYTHON_PATHS}", "/tmp") + .replace("{TRACING_PROXY_CA_CERT_PATH}", "/dev/null"); + + let final_config = finalize_nsjail_config(&config); + + // System dirs removed + for dir in &["/bin", "/lib", "/lib64", "/usr", "/etc"] { + assert!( + !final_config.contains(&format!("dst: \"{dir}\"")), + "{dir} should be stripped with overlay" + ); + } + + // Overlay root + assert!(final_config.contains("dst: \"/\"")); + assert!(final_config.contains("src: \"/tmp/jobXYZ/overlay_merged\"")); + + // Python-specific mounts preserved + assert!(final_config.contains("dst: \"/tmp/main.py\"")); + assert!(final_config.contains("dst: \"/tmp/wrapper.py\"")); + assert!(final_config.contains("dst: \"/tmp/args.json\"")); + assert!(final_config.contains("dst: \"/tmp/result.json\"")); + assert!(final_config.contains("dst: \"/dev/shm\"")); + assert!(final_config.contains("dst: \"/tmp\"")); + + // Python metadata + assert!(final_config.contains("name: \"python run script\"")); + assert!(final_config.contains("PYTHONPATH")); + } + + // --- End-to-end: volumes without snapshot (no overlay) --- + + #[test] + fn test_e2e_volumes_only_no_overlay() { + let script = "# volume: cache:/tmp/pip-cache\ndef main(): pass\n"; + let sandbox_config = parse_sandbox_config(script); + assert!(sandbox_config.snapshot.is_none()); + assert_eq!(sandbox_config.volumes.len(), 1); + + let mut setup = SandboxSetupState::default(); + setup.volume_mounts.insert( + "cache".to_string(), + (PathBuf::from("/tmp/job456/volumes/cache"), "/tmp/pip-cache".to_string()), + ); + + let sandbox_mounts = build_sandbox_mounts(&setup); + + // No overlay marker → system dirs should NOT be stripped + assert!(!sandbox_mounts.contains(OVERLAY_MARKER)); + + let raw_config = include_str!("../nsjail/run.python3.config.proto"); + let config = raw_config + .replace("{JOB_DIR}", "/tmp/job456") + .replace("{MAIN}", "main") + .replace("{CLONE_NEWUSER}", "true") + .replace("{SHARED_MOUNT}", &sandbox_mounts) + .replace("{SHARED_DEPENDENCIES}", "") + .replace("{PY_INSTALL_DIR}", "/usr/local") + .replace("{GLOBAL_SITE_PACKAGES}", "/usr/lib/python3/dist-packages") + .replace("{ADDITIONAL_PYTHON_PATHS}", "/tmp") + .replace("{TRACING_PROXY_CA_CERT_PATH}", "/dev/null"); + + let final_config = finalize_nsjail_config(&config); + + // System dirs preserved (no overlay) + assert!(final_config.contains("dst: \"/bin\""), "/bin should be kept"); + assert!(final_config.contains("dst: \"/lib\""), "/lib should be kept"); + assert!(final_config.contains("dst: \"/usr\""), "/usr should be kept"); + assert!(final_config.contains("dst: \"/etc\""), "/etc should be kept"); + + // Volume mount present + assert!(final_config.contains("dst: \"/tmp/pip-cache\"")); + assert!(final_config.contains("src: \"/tmp/job456/volumes/cache\"")); + } + + // --- End-to-end: no sandbox annotations at all (normal job) --- + + #[test] + fn test_e2e_no_annotations_passthrough() { + let script = "def main():\n return 42\n"; + let sandbox_config = parse_sandbox_config(script); + assert!(sandbox_config.snapshot.is_none()); + assert!(sandbox_config.volumes.is_empty()); + + // No sandbox setup → shared_mount is empty → config unchanged + let raw_config = include_str!("../nsjail/run.bash.config.proto"); + let config = raw_config + .replace("{JOB_DIR}", "/tmp/job789") + .replace("{CLONE_NEWUSER}", "true") + .replace("{SHARED_MOUNT}", "") + .replace("{TRACING_PROXY_CA_CERT_PATH}", "/dev/null"); + + let final_config = finalize_nsjail_config(&config); + + // Everything preserved as-is + assert!(final_config.contains("dst: \"/bin\"")); + assert!(final_config.contains("dst: \"/lib\"")); + assert!(final_config.contains("dst: \"/usr\"")); + assert!(final_config.contains("dst: \"/etc\"")); + assert!(final_config.contains("dst: \"/tmp\"")); + assert!(!final_config.contains(OVERLAY_MARKER)); + } + + // --- Verify mount block syntax is valid nsjail proto format --- + + #[test] + fn test_mount_block_syntax() { + let setup = SandboxSetupState { + overlay: Some(OverlayMount { + merged: PathBuf::from("/job/merged"), + upper: PathBuf::from("/job/upper"), + work: PathBuf::from("/job/work"), + }), + volume_mounts: HashMap::from([( + "vol".to_string(), + (PathBuf::from("/job/volumes/vol"), "/mnt/vol".to_string()), + )]), + }; + let mounts = build_sandbox_mounts(&setup); + + // Each mount block should have proper nsjail proto syntax + for block in mounts.split("mount {").skip(1) { + assert!(block.contains('}'), "Every mount block must close"); + assert!(block.contains("dst: \""), "Every mount needs a dst"); + assert!(block.contains("is_bind: true"), "Sandbox mounts are bind mounts"); + assert!(block.contains("rw: true"), "Sandbox mounts are read-write"); + } + } +} diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 547075933a..4f2fff1687 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -4147,7 +4147,7 @@ pub async fn run_language_executor( job.id ); - let shared_mount = if job.same_worker && job.script_lang != Some(ScriptLang::Deno) { + let mut shared_mount = if job.same_worker && job.script_lang != Some(ScriptLang::Deno) { let folder = if job.script_lang == Some(ScriptLang::Go) { "/go" } else { @@ -4167,7 +4167,45 @@ mount {{ "".to_string() }; - // println!("handle lang job {:?}", SystemTime::now()); + let sandbox_config = windmill_common::sandbox::parse_sandbox_config(&code); + let sandbox_setup = if sandbox_config.snapshot.is_some() || !sandbox_config.volumes.is_empty() { + let db = conn.as_sql().ok_or_else(|| { + Error::ExecutionErr("Sandbox features require SQL connection".to_string()) + })?; + let mut setup = crate::sandbox_setup::SandboxSetupState::default(); + + if let Some(ref snap) = sandbox_config.snapshot { + let snapshot_path = crate::sandbox_setup::ensure_snapshot_cached( + &job.workspace_id, &snap.name, &snap.tag, db, + ) + .await?; + setup.overlay = + Some(crate::sandbox_setup::mount_overlay(&snapshot_path, job_dir).await?); + } + + for (vol_name, mount_path) in &sandbox_config.volumes { + let local_dir = format!("{job_dir}/volumes/{vol_name}"); + std::fs::create_dir_all(&local_dir).map_err(|e| { + Error::ExecutionErr(format!("Failed to create volume dir: {e}")) + })?; + crate::sandbox_setup::download_volume( + &job.workspace_id, + vol_name, + std::path::Path::new(&local_dir), + db, + ) + .await?; + setup.volume_mounts.insert( + vol_name.clone(), + (std::path::PathBuf::from(&local_dir), mount_path.clone()), + ); + } + + shared_mount.push_str(&crate::sandbox_setup::build_sandbox_mounts(&setup)); + Some(setup) + } else { + None + }; let envs = build_envs(envs.as_ref())?; @@ -4623,7 +4661,26 @@ mount {{ &lang_str, job.id ); - // println!("handled job: {:?}", SystemTime::now()); + + if let Some(ref setup) = sandbox_setup { + if let Some(db) = conn.as_sql() { + for (vol_name, (local_dir, _)) in &setup.volume_mounts { + if let Err(e) = crate::sandbox_setup::upload_volume( + &job.workspace_id, + vol_name, + local_dir, + db, + ) + .await + { + tracing::error!("Failed to upload volume {vol_name}: {e}"); + } + } + } + if let Some(ref overlay) = setup.overlay { + crate::sandbox_setup::unmount_overlay(overlay).await.ok(); + } + } result }