diff --git a/backend/.sqlx/query-0df1c23ac429b5807d42e1da5ee2e2176a8a774f107bece5804cc7b221365c89.json b/backend/.sqlx/query-0df1c23ac429b5807d42e1da5ee2e2176a8a774f107bece5804cc7b221365c89.json new file mode 100644 index 0000000000..cc5e9a8d45 --- /dev/null +++ b/backend/.sqlx/query-0df1c23ac429b5807d42e1da5ee2e2176a8a774f107bece5804cc7b221365c89.json @@ -0,0 +1,58 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH RECURSIVE job_tree AS (\n SELECT id, tag FROM v2_job WHERE id = $2 AND workspace_id = $1\n UNION\n SELECT j.id, j.tag FROM v2_job j JOIN job_tree t ON j.parent_job = t.id\n WHERE j.workspace_id = $1\n )\n SELECT\n a.path,\n a.kind AS \"kind!: windmill_common::assets::AssetKind\",\n -- Several jobs of the tree touch one asset, each recording its own\n -- access. A job that recorded none contributes nothing rather than\n -- erasing a sibling's, so an all-null group is the only unknown one.\n -- Grouping here, not in Rust, is what makes LIMIT count assets: the\n -- retention keeps up to ten job rows per asset.\n COALESCE(bool_or(a.usage_access_type IN ('r', 'rw')), false) AS \"any_read!\",\n COALESCE(bool_or(a.usage_access_type IN ('w', 'rw')), false) AS \"any_write!\"\n FROM asset a JOIN job_tree t ON a.usage_path = t.id::text\n WHERE a.workspace_id = $1 AND a.usage_kind = 'job'\n AND ($3::text[] IS NULL OR t.tag = ANY($3))\n GROUP BY a.path, a.kind\n ORDER BY a.path, a.kind\n LIMIT $4", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "kind!: windmill_common::assets::AssetKind", + "type_info": { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume", + "dbt" + ] + } + } + } + }, + { + "ordinal": 2, + "name": "any_read!", + "type_info": "Bool" + }, + { + "ordinal": 3, + "name": "any_write!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Uuid", + "TextArray", + "Int8" + ] + }, + "nullable": [ + false, + false, + null, + null + ] + }, + "hash": "0df1c23ac429b5807d42e1da5ee2e2176a8a774f107bece5804cc7b221365c89" +} diff --git a/backend/windmill-api-integration-tests/tests/assets.rs b/backend/windmill-api-integration-tests/tests/assets.rs new file mode 100644 index 0000000000..e1e7645e52 --- /dev/null +++ b/backend/windmill-api-integration-tests/tests/assets.rs @@ -0,0 +1,248 @@ +use serde_json::{json, Value}; +use sqlx::{Pool, Postgres}; +use uuid::Uuid; +use windmill_test_utils::*; + +fn client() -> reqwest::Client { + reqwest::Client::new() +} + +fn bearer(builder: reqwest::RequestBuilder, token: &str) -> reqwest::RequestBuilder { + builder.header("Authorization", format!("Bearer {token}")) +} + +async fn insert_job(db: &Pool, parent: Option, tag: &str) -> anyhow::Result { + let id = Uuid::new_v4(); + sqlx::query( + "INSERT INTO v2_job (id, workspace_id, tag, created_by, permissioned_as, \ + permissioned_as_email, kind, parent_job, same_worker, visible_to_owner) \ + VALUES ($1, 'test-workspace', $3, 'test-user', 'u/test-user', \ + 'test@windmill.dev', 'script', $2, false, true)", + ) + .bind(id) + .bind(parent) + .bind(tag) + .execute(db) + .await?; + Ok(id) +} + +/// `access_type` is `None` for a detection that could not tell read from write +/// — how a resource passed in a job's arguments is recorded. +async fn insert_job_asset( + db: &Pool, + job: Uuid, + path: &str, + access_type: Option<&str>, +) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind) \ + VALUES ('test-workspace', $1, 's3object', $2::text::asset_access_type, $3, 'job')", + ) + .bind(path) + .bind(access_type) + .bind(job.to_string()) + .execute(db) + .await?; + Ok(()) +} + +/// A run reports what its whole job tree touched: runtime detection records +/// against the job that did the operation, which for a flow step or a +/// workflow-as-code task is never the job the user opened. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_list_run_assets_covers_child_jobs(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let ws = format!("http://localhost:{port}/api/w/test-workspace"); + + let parent = insert_job(&db, None, "other").await?; + let child = insert_job(&db, Some(parent), "other").await?; + let unrelated = insert_job(&db, None, "other").await?; + + insert_job_asset(&db, parent, "/data/shared.json", Some("r")).await?; + insert_job_asset(&db, child, "/data/shared.json", Some("w")).await?; + insert_job_asset(&db, child, "/data/child_only.json", Some("w")).await?; + insert_job_asset(&db, parent, "/data/from_args.json", None).await?; + insert_job_asset(&db, child, "/data/from_args.json", Some("w")).await?; + insert_job_asset(&db, unrelated, "/data/unrelated.json", Some("w")).await?; + + let resp = bearer( + client().get(format!("{ws}/jobs/run_assets/{parent}")), + "SECRET_TOKEN", + ) + .send() + .await?; + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await?; + assert_eq!(body["truncated"], json!(false)); + assert_eq!( + body["assets"], + json!([ + { "path": "/data/child_only.json", "kind": "s3object", "access_type": "w" }, + // The parent recorded no access type for this one; that must not erase + // the child's. + { "path": "/data/from_args.json", "kind": "s3object", "access_type": "w" }, + { "path": "/data/shared.json", "kind": "s3object", "access_type": "rw" }, + ]), + "parent should report its own and its child's assets, with access types merged" + ); + + let resp = bearer( + client().get(format!("{ws}/jobs/run_assets/{child}")), + "SECRET_TOKEN", + ) + .send() + .await?; + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await?; + assert_eq!( + body["assets"], + json!([ + { "path": "/data/child_only.json", "kind": "s3object", "access_type": "w" }, + { "path": "/data/from_args.json", "kind": "s3object", "access_type": "w" }, + { "path": "/data/shared.json", "kind": "s3object", "access_type": "w" }, + ]), + "a child should report only what it touched itself" + ); + + // `asset` has no RLS of its own, so the job read gate is the only thing + // standing between another member and these paths. + let resp = bearer( + client().get(format!("{ws}/jobs/run_assets/{parent}")), + "SECRET_TOKEN_2", + ) + .send() + .await?; + assert_eq!( + resp.status().as_u16(), + 403, + "a member who cannot read the run must not read its assets" + ); + + // ...and a share link is what lets that same member in. The tree is walked + // outside the caller's RLS precisely so this works, since the token grants + // access their own permissions do not. + let view_token: String = bearer( + client().get(format!("{ws}/jobs/job_view_token/{parent}")), + "SECRET_TOKEN", + ) + .send() + .await? + .text() + .await?; + let resp = bearer( + client().get(format!("{ws}/jobs/run_assets/{parent}")), + "SECRET_TOKEN_2", + ) + .header("X-View-Token", view_token) + .send() + .await?; + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await?; + assert_eq!( + body["assets"].as_array().map(|a| a.len()), + Some(3), + "a share-link viewer should see the whole tree's assets" + ); + + Ok(()) +} + +/// The read gate only checks the root job's tag, so the walk has to keep a +/// tag-scoped token out of descendants outside its scope — without hiding the +/// ones below them, which the token could have asked for directly. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_list_run_assets_scopes_descendants_by_tag(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let ws = format!("http://localhost:{port}/api/w/test-workspace"); + + sqlx::query( + "INSERT INTO token (token_hash, token_prefix, token, email, label, super_admin, scopes) \ + VALUES (encode(sha256('TAG_TOKEN'::bytea), 'hex'), 'TAG_TOK', 'TAG_TOKEN', \ + 'test@windmill.dev', 'tag-only', true, ARRAY['if_jobs:filter_tags:other'])", + ) + .execute(&db) + .await?; + + let parent = insert_job(&db, None, "other").await?; + let in_scope = insert_job(&db, Some(parent), "other").await?; + let out_of_scope = insert_job(&db, Some(parent), "deno").await?; + let below_out_of_scope = insert_job(&db, Some(out_of_scope), "other").await?; + insert_job_asset(&db, in_scope, "/data/in_scope.json", Some("w")).await?; + insert_job_asset(&db, out_of_scope, "/data/out_of_scope.json", Some("w")).await?; + insert_job_asset(&db, below_out_of_scope, "/data/nested.json", Some("w")).await?; + + let resp = bearer( + client().get(format!("{ws}/jobs/run_assets/{parent}")), + "TAG_TOKEN", + ) + .send() + .await?; + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await?; + assert_eq!( + body["assets"], + json!([ + { "path": "/data/in_scope.json", "kind": "s3object", "access_type": "w" }, + { "path": "/data/nested.json", "kind": "s3object", "access_type": "w" }, + ]), + "a tag-scoped token must not read assets of descendants outside its tags, \ + but must still reach in-scope jobs below them" + ); + + Ok(()) +} + +/// A fan-out run can touch more assets than one response should carry, and the +/// cut must be reported rather than served as if it were the whole list. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_list_run_assets_caps_the_list(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let ws = format!("http://localhost:{port}/api/w/test-workspace"); + + // Three child jobs touching the same 1200 assets: the cap counts assets, and + // the retention allows ten job rows per asset, so a row-counted cap would cut + // this at a third of the list. + let parent = insert_job(&db, None, "other").await?; + for _ in 0..3 { + let child = insert_job(&db, Some(parent), "other").await?; + sqlx::query( + "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind) \ + SELECT 'test-workspace', '/out/' || lpad(g::text, 6, '0') || '.json', 's3object', 'w', \ + $1, 'job' FROM generate_series(1, 1200) g", + ) + .bind(child.to_string()) + .execute(&db) + .await?; + } + + let resp = bearer( + client().get(format!("{ws}/jobs/run_assets/{parent}")), + "SECRET_TOKEN", + ) + .send() + .await?; + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await?; + assert_eq!(body["truncated"], json!(true)); + let assets = body["assets"].as_array().expect("assets array"); + assert_eq!(assets.len(), 1000); + assert_eq!( + assets[0], + json!({ "path": "/out/000001.json", "kind": "s3object", "access_type": "w" }), + "the cap keeps the head of the ordered list, with its access type merged" + ); + // Three jobs touched each asset, so a cap counting rows rather than assets + // would fill the list with repeats and stop around /out/000334.json. + assert_eq!( + assets[999]["path"], "/out/001000.json", + "the cap counts assets, not asset rows" + ); + Ok(()) +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index ed9479253e..f8579f9aa4 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -23847,6 +23847,55 @@ paths: items: $ref: "#/components/schemas/AssetProgress" + /w/{workspace}/jobs/run_assets/{id}: + get: + summary: List the assets a run touched at runtime + description: > + Assets detected while the run executed (SDK S3 calls, resources passed as + arguments), aggregated over the job and all of its child jobs so a flow or + workflow-as-code run reports what its steps and tasks touched. Authorized + through the job, the same gate as `run_progress`. Recording is asynchronous, + so an asset can take a few minutes after the run to appear, and only the most + recent runs that touched an asset keep that record. A run that fans out can + touch more assets than one response should carry, so the list is capped and + `truncated` says when it was cut. + operationId: listRunAssets + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: id + in: path + required: true + description: The job whose runtime assets to read + schema: + type: string + format: uuid + responses: + "200": + description: assets this run and its child jobs touched + content: + application/json: + schema: + type: object + required: [assets, truncated] + properties: + truncated: + type: boolean + description: whether the run touched more assets than are listed + assets: + type: array + items: + type: object + required: [path, kind] + properties: + path: + type: string + kind: + $ref: "#/components/schemas/AssetKind" + access_type: + $ref: "#/components/schemas/AssetUsageAccessType" + /w/{workspace}/assets/partitions_in_range: get: summary: List expected partitions of a ducklake asset in a date range with their materialization status (enterprise) diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index b79e0b8b14..584c44bbec 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -25,6 +25,7 @@ use std::time::Instant; use tokio::io::AsyncReadExt; use tower::ServiceBuilder; use url::Url; +use windmill_common::assets::AssetUsageAccessType; #[cfg(all(feature = "enterprise", feature = "instance_smtp"))] use windmill_common::auth::is_super_admin_email; use windmill_common::auth::TOKEN_PREFIX_LEN; @@ -136,6 +137,7 @@ pub fn workspaced_service() -> Router { Router::new() .route("/run_progress/{id}", get(get_run_progress)) + .route("/run_assets/{id}", get(list_run_assets)) .route("/dbt_graph/{id}", get(get_dbt_run_graph)) .route("/dbt_resumable/{id}", get(get_dbt_resumable)) .route( @@ -1188,6 +1190,121 @@ async fn get_run_progress( Ok(Json(rows)) } +#[derive(Serialize)] +struct RunAsset { + path: String, + kind: windmill_common::assets::AssetKind, + #[serde(skip_serializing_if = "Option::is_none")] + access_type: Option, +} + +#[derive(Serialize)] +struct RunAssets { + assets: Vec, + truncated: bool, +} + +/// A fan-out run — a forloop writing one object per iteration — touches as many +/// assets as it has steps, and the whole list would land in one response and one +/// list in the browser. Cap it, and say so rather than serving a prefix that +/// reads like the whole answer. +const RUN_ASSETS_CAP: usize = 1000; + +/// Assets a run touched, as recorded by runtime detection, aggregated over the +/// whole job tree: the recorder attributes an asset to the job that performed +/// the operation, which for a flow step or a workflow-as-code task is not the +/// job the user opened. Lives with the job routes for the same reason +/// `run_progress` does — `asset` has no RLS, and `require_job_read_access` is +/// what applies the view token, a scoped token's tag filter and the app-embed +/// cutoff. +async fn list_run_assets( + authed: ApiAuthed, + OptViewToken(view_token): OptViewToken, + Extension(db): Extension, + Extension(user_db): Extension, + Path((w_id, job_id)): Path<(String, Uuid)>, +) -> error::JsonResult { + let created_by = sqlx::query_scalar!( + "SELECT created_by FROM v2_job WHERE id = $1 AND workspace_id = $2", + job_id, + &w_id + ) + .fetch_optional(&db) + .await?; + let Some(created_by) = created_by else { + return Ok(Json(RunAssets { assets: vec![], truncated: false })); + }; + require_job_read_access( + &db, + &user_db, + &authed, + &w_id, + &job_id, + &created_by, + view_token.as_deref(), + ) + .await?; + + // Walked on `db`, like the flow tree is: the gate above is what authorizes the + // run, and it grants access the caller's own RLS does not have — a view token, + // or a job the caller launched that runs as someone else. Re-filtering the tree + // through `user_db` would drop exactly those, and hand a share-link viewer the + // empty tab this endpoint exists to fix. The tag scope is the one restriction + // that must still hold per job, since the gate only checked the root. It gates + // which jobs' assets are read, not which are walked through: the gate admits a + // job on its own tag, so an out-of-scope job in the middle of the tree must not + // hide a descendant the caller could have asked for directly. + let scope_tags = get_scope_tags(&authed).map(|v| v.iter().map(|s| s.to_string()).collect_vec()); + let rows = sqlx::query!( + r#"WITH RECURSIVE job_tree AS ( + SELECT id, tag FROM v2_job WHERE id = $2 AND workspace_id = $1 + UNION + SELECT j.id, j.tag FROM v2_job j JOIN job_tree t ON j.parent_job = t.id + WHERE j.workspace_id = $1 + ) + SELECT + a.path, + a.kind AS "kind!: windmill_common::assets::AssetKind", + -- Several jobs of the tree touch one asset, each recording its own + -- access. A job that recorded none contributes nothing rather than + -- erasing a sibling's, so an all-null group is the only unknown one. + -- Grouping here, not in Rust, is what makes LIMIT count assets: the + -- retention keeps up to ten job rows per asset. + COALESCE(bool_or(a.usage_access_type IN ('r', 'rw')), false) AS "any_read!", + COALESCE(bool_or(a.usage_access_type IN ('w', 'rw')), false) AS "any_write!" + FROM asset a JOIN job_tree t ON a.usage_path = t.id::text + WHERE a.workspace_id = $1 AND a.usage_kind = 'job' + AND ($3::text[] IS NULL OR t.tag = ANY($3)) + GROUP BY a.path, a.kind + ORDER BY a.path, a.kind + LIMIT $4"#, + w_id, + job_id, + scope_tags.as_deref(), + // One asset past the cap is how the response learns it was cut. + RUN_ASSETS_CAP as i64 + 1 + ) + .fetch_all(&db) + .await?; + + let truncated = rows.len() > RUN_ASSETS_CAP; + let assets = rows + .into_iter() + .take(RUN_ASSETS_CAP) + .map(|row| RunAsset { + path: row.path, + kind: row.kind, + access_type: match (row.any_read, row.any_write) { + (true, true) => Some(AssetUsageAccessType::RW), + (true, false) => Some(AssetUsageAccessType::R), + (false, true) => Some(AssetUsageAccessType::W), + (false, false) => None, + }, + }) + .collect(); + Ok(Json(RunAssets { assets, truncated })) +} + async fn get_flow_job_debug_info( OptViewToken(view_token): OptViewToken, OptAuthed(opt_authed): OptAuthed, @@ -1654,20 +1771,21 @@ async fn require_job_within_run_scope( .await?; let runs_app = build_scope_path_predicate(authed, "apps", "run"); - let in_scope = chain.iter().any(|job| { - match (job.runnable_path.as_deref(), job.scope_kind.as_deref()) { - (Some(runnable_path), Some(kind)) - if windmill_api_auth::scopes::run_confinement_admits( - &confinement, - kind, - runnable_path, - ) => - { - true - } - _ => job.launched_by_app.as_deref().is_some_and(&runs_app), - } - }); + let in_scope = + chain.iter().any( + |job| match (job.runnable_path.as_deref(), job.scope_kind.as_deref()) { + (Some(runnable_path), Some(kind)) + if windmill_api_auth::scopes::run_confinement_admits( + &confinement, + kind, + runnable_path, + ) => + { + true + } + _ => job.launched_by_app.as_deref().is_some_and(&runs_app), + }, + ); if in_scope { Ok(()) diff --git a/frontend/src/lib/components/assets/JobAssetsViewer.svelte b/frontend/src/lib/components/assets/JobAssetsViewer.svelte index 10e2f011f9..45bc4cd70f 100644 --- a/frontend/src/lib/components/assets/JobAssetsViewer.svelte +++ b/frontend/src/lib/components/assets/JobAssetsViewer.svelte @@ -1,13 +1,15 @@ -{#if assets.value && assets.value.length > 0} +{#if assets.status === 'idle' || assets.status === 'loading'} + +{:else if assets.value && assets.value.assets.length > 0}
    - {#each assets.value ?? [] as asset} -
  • + {#each assets.value.assets as asset} +
  • {asset.path} @@ -90,17 +139,26 @@ })}
    - + {#if asset.access_type} + {formatAssetAccessType(asset.access_type)} + {/if} +
    • {/each}
+ {#if assets.value.truncated} +
+ This run touched more assets than are listed here. +
+ {/if} {:else} -
No assets found
+
+ No assets found + + Assets detected while a run executes are recorded asynchronously, and only the most recent + runs that touched an asset keep that record. + +
{/if}