starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-12 16:02:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: step-level job_token_scopes for flow steps and agent tools

Browse Source 
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014FHGE3ynoAu6yrgeg4kLwL
This commit is contained in:
Ruben Fiszel
2026-10-02 19:44:39 +02:00
co-authored by Claude Opus 5.5
parent fc9fec6e0b
commit d21809dad4
18 changed files with 150 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
9a8cd3db05b6c0ac69b05d97ae5355d98b812e0f
d57d2d7c0613028ae14f0764b8bf8ce259a06417
backend/tests/flow_engine_parity.rs
+1
View File
@@ -50,6 +50,7 @@ fn flow_module(id: &str, value: FlowModuleValue) -> FlowModule {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
}
}
backend/tests/job_token_scopes.rs
+39 -23
View File
@@ -175,45 +175,61 @@ async fn test_restricted_job_token_is_confined(db: Pool<Postgres>) -> anyhow::Re
#[sqlx::test(fixtures("base"))]
async fn test_flow_steps_inherit_the_flow_restriction(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let value = json!({ "modules": [{ "id": "a", "value": { "type": "identity" } }] });
let value = json!({ "modules": [
{ "id": "a", "value": { "type": "identity" },
"job_token_scopes": ["oidc:write", "variables:read"] },
{ "id": "b", "value": { "type": "identity" } },
] });
sqlx::query(
"INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, edited_at,
schema, extra_perms, versions, job_token_scopes)
VALUES ('test-workspace', 'u/test-user/restricted', '', '', $1, 'test-user', now(), '{}',
'{}', '{7171}', '{oidc:write}')",
schema, extra_perms, versions)
VALUES ('test-workspace', 'u/test-user/agent_flow', '', '', $1, 'test-user', now(), '{}',
'{}', '{7171}')",
)
.bind(&value)
.execute(&db)
.await?;
sqlx::query(
"INSERT INTO flow_version (id, workspace_id, path, value, schema, created_by)
VALUES (7171, 'test-workspace', 'u/test-user/restricted', $1, '{}', 'test-user')",
VALUES (7171, 'test-workspace', 'u/test-user/agent_flow', $1, '{}', 'test-user')",
)
.bind(&value)
.execute(&db)
.await?;
let server = ApiServer::start(db.clone()).await?;
let flow = RunJob::from(JobPayload::Flow {
path: "u/test-user/restricted".to_string(),
dedicated_worker: None,
apply_preprocessor: false,
version: 7171,
labels: None,
job_token_scopes: Some(vec!["oidc:write".to_string()]),
})
.run_until_complete(&db, false, server.addr.port())
.await;
let scopes = |v: &[&str]| Some(v.iter().map(|s| s.to_string()).collect::<Vec<_>>());
// A step's own setting narrows the flow's restriction and never widens it; in an
// unrestricted flow it restricts that step alone.
for (flow_scopes, expected_a, expected_b) in [
(scopes(&["oidc:write"]), scopes(&["oidc:write"]), scopes(&["oidc:write"])),
(None, scopes(&["oidc:write", "variables:read"]), None),
] {
let flow = RunJob::from(JobPayload::Flow {
path: "u/test-user/agent_flow".to_string(),
dedicated_worker: None,
apply_preprocessor: false,
version: 7171,
labels: None,
job_token_scopes: flow_scopes.clone(),
})
.run_until_complete(&db, false, server.addr.port())
.await;
let step: Option<Vec<String>> =
sqlx::query_scalar(
"SELECT p.job_token_scopes FROM v2_job j JOIN job_perms p ON p.job_id = j.id
WHERE j.parent_job = $1",
let steps: Vec<(String, Option<Vec<String>>)> = sqlx::query_as(
"SELECT j.flow_step_id, p.job_token_scopes FROM v2_job j
JOIN job_perms p ON p.job_id = j.id
WHERE j.parent_job = $1 ORDER BY j.flow_step_id",
)
.bind(flow.id)
.fetch_one(&db)
.await?;
assert_eq!(step, Some(vec!["oidc:write".to_string()]));
.bind(flow.id)
.fetch_all(&db)
.await?;
assert_eq!(
steps,
vec![("a".to_string(), expected_a), ("b".to_string(), expected_b)],
"flow scopes {flow_scopes:?}"
);
}
Ok(())
}
backend/tests/worker.rs
+8
View File
@@ -212,6 +212,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
FlowModule {
id: "b".to_string(),
@@ -260,6 +261,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
}],
modules_node: None,
}
@@ -282,6 +284,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
],
same_worker: false,
@@ -398,6 +401,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
FlowModule {
id: "b".to_string(),
@@ -455,6 +459,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
FlowModule {
id: "e".to_string(),
@@ -498,6 +503,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
],
modules_node: None,
@@ -520,6 +526,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
FlowModule {
id: "c".to_string(),
@@ -569,6 +576,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
],
same_worker: true,
backend/windmill-api-flows/src/flows.rs
+20 -2
View File
@@ -809,11 +809,18 @@ async fn create_flow(
.and_then(|scopes| scopes.as_deref())
.map(windmill_common::scopes::validate_job_token_scopes)
.transpose()?;
if job_token_scopes.is_some() {
let restricts_steps = windmill_common::scopes::validate_flow_step_job_token_scopes(
&serde_json::from_str::<windmill_common::flows::FlowValue>(nf.value.get())
.map_err(|e| windmill_common::error::Error::BadRequest(e.to_string()))?,
)?;
if job_token_scopes.is_some() || restricts_steps {
windmill_common::min_version::MIN_VERSION_SUPPORTS_JOB_TOKEN_SCOPES
.assert()
.await?;
}
if restricts_steps {
windmill_common::feature_usage::log_feature_usage("job_token_scopes", "deploy", "step:set");
}
let resolved_on_behalf_of = windmill_common::resolve_on_behalf_of(
nf.on_behalf_of_email.as_deref(),
nf.on_behalf_of.as_deref(),
@@ -1391,11 +1398,18 @@ async fn update_flow(
.and_then(|scopes| scopes.as_deref())
.map(windmill_common::scopes::validate_job_token_scopes)
.transpose()?;
if job_token_scopes.is_some() {
let restricts_steps = windmill_common::scopes::validate_flow_step_job_token_scopes(
&serde_json::from_str::<windmill_common::flows::FlowValue>(nf.value.get())
.map_err(|e| windmill_common::error::Error::BadRequest(e.to_string()))?,
)?;
if job_token_scopes.is_some() || restricts_steps {
windmill_common::min_version::MIN_VERSION_SUPPORTS_JOB_TOKEN_SCOPES
.assert()
.await?;
}
if restricts_steps {
windmill_common::feature_usage::log_feature_usage("job_token_scopes", "deploy", "step:set");
}
let resolved_on_behalf_of = windmill_common::resolve_on_behalf_of(
nf.on_behalf_of_email.as_deref(),
nf.on_behalf_of.as_deref(),
@@ -2373,6 +2387,7 @@ mod tests {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
FlowModule {
id: "b".to_string(),
@@ -2408,6 +2423,7 @@ mod tests {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
FlowModule {
id: "c".to_string(),
@@ -2443,6 +2459,7 @@ mod tests {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
},
],
failure_module: Some(Box::new(FlowModule {
@@ -2477,6 +2494,7 @@ mod tests {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
})),
preprocessor_module: None,
same_worker: false,
backend/windmill-common/src/scopes.rs
+24
View File
@@ -618,6 +618,30 @@ pub fn intersect_job_token_scopes(
}
}
/// Validates the `job_token_scopes` of every step and agent tool of a flow, returning whether
/// any of them sets one.
pub fn validate_flow_step_job_token_scopes(value: &crate::flows::FlowValue) -> Result<bool> {
let mut any = false;
let mut check = |module: &crate::flows::FlowModule| -> anyhow::Result<()> {
if let Some(scopes) = &module.job_token_scopes {
any = true;
validate_job_token_scopes(scopes)
.map_err(|e| anyhow::anyhow!("step {}: {e}", module.id))?;
}
Ok(())
};
let extra: Vec<crate::flows::FlowModule> = value
.failure_module
.iter()
.chain(value.preprocessor_module.iter())
.map(|m| (**m).clone())
.collect();
crate::flows::FlowModule::traverse_modules(&value.modules, &mut check)
.and_then(|()| crate::flows::FlowModule::traverse_modules(&extra, &mut check))
.map_err(|e| Error::BadRequest(e.to_string()))?;
Ok(any)
}
/// Counts a deploy of a script or flow that restricts its job token, keyed by which shape
/// of restriction it picked, so take-up of the presets can be told from custom lists.
pub fn log_job_token_scopes_deploy(kind: &'static str, scopes: Option<&[String]>) {
backend/windmill-types/src/flows.rs
+9
View File
@@ -635,6 +635,9 @@ pub struct FlowModule {
pub pass_flow_input_directly: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debouncing: Option<DebouncingSettings>,
/// Caps the token of the jobs this step runs, on top of the flow's own restriction.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub job_token_scopes: Option<Vec<String>>,
}
#[derive(Deserialize, Serialize, Debug, Clone)]
@@ -922,6 +925,9 @@ pub struct AgentTool {
/// Overrides the description auto-derived from the underlying runnable.
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
/// Caps the token of this tool's jobs, on top of the agent step's own restriction.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub job_token_scopes: Option<Vec<String>>,
pub value: ToolValue,
}
@@ -935,6 +941,7 @@ impl AgentTool {
self.id = flow_module.id;
self.summary = flow_module.summary;
self.job_token_scopes = flow_module.job_token_scopes;
self.value = ToolValue::FlowModule(module_value);
}
}
@@ -947,6 +954,7 @@ impl From<&AgentTool> for Option<FlowModule> {
id: tool.id.clone(),
value: to_raw_value(module_value),
summary: tool.summary.clone(),
job_token_scopes: tool.job_token_scopes.clone(),
..Default::default()
}),
ToolValue::Mcp(_) => None,
@@ -1351,6 +1359,7 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
apply_preprocessor: None,
pass_flow_input_directly: None,
debouncing: None,
job_token_scopes: None,
});
}
}
backend/windmill-worker/src/ai/tools.rs
+7 -2
View File
@@ -599,8 +599,13 @@ async fn enqueue_windmill_tool(
None,
None,
None,
// A tool never holds a wider token than the agent calling it.
ctx.job.job_token_scopes.as_deref(),
// A tool never holds a wider token than the agent calling it, and its own setting
// narrows that further.
windmill_common::scopes::intersect_job_token_scopes(
ctx.job.job_token_scopes.as_deref(),
tool_module.job_token_scopes.as_deref(),
)
.as_deref(),
)
.await?;
backend/windmill-worker/src/ai_executor.rs
+5
View File
@@ -2745,6 +2745,7 @@ mod tests {
let mut its = HashMap::new();
its.insert(key.to_string(), js(expr));
AgentTool {
job_token_scopes: None,
id: id.to_string(),
summary: None,
description: None,
@@ -2772,6 +2773,7 @@ mod tests {
script_tool("a", "x", "authoring_flow_expr"),
script_tool("b", "y", "keep_me"),
AgentTool {
job_token_scopes: None,
id: "m".to_string(),
summary: None,
description: None,
@@ -2819,6 +2821,7 @@ mod tests {
fn narrow_roster_keeps_the_entries_a_run_named() {
fn named(id: &str, summary: &str) -> AgentTool {
AgentTool {
job_token_scopes: None,
id: id.to_string(),
summary: Some(summary.to_string()),
description: None,
@@ -2834,6 +2837,7 @@ mod tests {
}
fn mcp(id: &str, summary: &str, path: &str) -> AgentTool {
AgentTool {
job_token_scopes: None,
id: id.to_string(),
summary: Some(summary.to_string()),
description: None,
@@ -2846,6 +2850,7 @@ mod tests {
}
fn websearch(id: &str, summary: Option<&str>) -> AgentTool {
AgentTool {
job_token_scopes: None,
id: id.to_string(),
summary: summary.map(str::to_string),
description: None,
backend/windmill-worker/src/worker.rs
-2
View File
@@ -16,8 +16,6 @@ use tokio::time::timeout;
// Re-export proxy env-var snapshots so callers (including EE modules)
// can keep importing them via `crate::{NO_PROXY, HTTP_PROXY, HTTPS_PROXY}`.
use windmill_common::client::AuthedClient;
use windmill_common::db::UserDbWithAuthed;
use windmill_common::get_latest_deployed_hash_for_path;
use windmill_common::jobs::InlineScriptTarget;
use windmill_common::jobs::RunInlineScriptFnParams;
use windmill_common::jobs::WorkerInternalServerInlineUtils;
backend/windmill-worker/src/worker_flow.rs
+7 -2
View File
@@ -4751,8 +4751,13 @@ async fn push_next_flow_job(
end_user_email,
None,
None,
// A step never holds a wider token than the flow running it.
flow_job.job_token_scopes.as_deref(),
// A step never holds a wider token than the flow running it, and its own setting
// narrows that further.
windmill_common::scopes::intersect_job_token_scopes(
flow_job.job_token_scopes.as_deref(),
module.job_token_scopes.as_deref(),
)
.as_deref(),
)
.warn_after_seconds(2)
.await?;
cli/src/guidance/skills.gen.ts
Generated
+1 -1
View File
File diff suppressed because one or more lines are too long
frontend/src/lib/components/JobTokenScopesSetting.svelte
+8 -3
View File
@@ -7,7 +7,7 @@
interface Props {
/** `null`/`undefined`: the job token has the full permissions of the identity it runs as. */
value: string[] | null | undefined
kind: 'script' | 'flow'
kind: 'script' | 'flow' | 'step'
}
let { value = $bindable(), kind }: Props = $props()
@@ -62,8 +62,13 @@
/>
{/if}
<p class="text-2xs text-secondary">
The <code>WM_TOKEN</code> of every job of this {kind} can only do what these scopes allow. Jobs
it starts, {kind === 'flow' ? 'its steps, ' : ''}and AI agent tools inherit the restriction.
{#if kind === 'step'}
The <code>WM_TOKEN</code> of every job this step runs can only do what these scopes allow, on top
of the flow's own restriction. Loop and branch bodies, and AI agent tools, inherit it.
{:else}
The <code>WM_TOKEN</code> of every job of this {kind} can only do what these scopes allow. Jobs
it starts, {kind === 'flow' ? 'its steps, ' : ''}and AI agent tools inherit the restriction.
{/if}
Variables and resources passed as <code>$var:</code>/<code>$res:</code> inputs, an AI agent's provider
resource, relative imports and object storage each need a read scope, and script state
(<code>getState</code>/<code>setState</code>) needs <code>resources:write</code>. Write scopes on scripts,
frontend/src/lib/components/flows/content/FlowRunSettings.svelte
+5
View File
@@ -28,6 +28,7 @@
import FlowModuleDebounce from './FlowModuleDebounce.svelte'
import FlowModuleMock from './FlowModuleMock.svelte'
import JobTokenScopesSetting from '$lib/components/JobTokenScopesSetting.svelte'
import WorkspaceScriptSettingInfo from './WorkspaceScriptSettingInfo.svelte'
import { slideDynamic } from '$lib/transitions'
@@ -382,4 +383,8 @@
{/if}
</section>
{/if}
<section class="flex flex-col gap-3" data-setting="job-token">
{@render sectionHeader('Job token')}
<JobTokenScopesSetting bind:value={flowModule.job_token_scopes} kind="step" />
</section>
</div>
openflow.openapi.yaml
+12
View File
@@ -376,6 +376,12 @@ components:
continue_on_error:
type: boolean
description: If true, flow continues even if this step fails
job_token_scopes:
type: array
nullable: true
items:
type: string
description: "Scopes the token of this step's jobs is restricted to (domain:action[:resource], e.g. oidc:write), on top of the flow's own restriction: a step can only narrow it. Applies to everything the step runs (loop and branch bodies, AI agent tools)."
retry:
description: Retry configuration if this step fails
$ref: '#/components/schemas/Retry'
@@ -999,6 +1005,12 @@ components:
description:
type: string
description: Free-text description of the tool given to the AI to decide when and how to call it. Overrides the description auto-derived from the underlying script.
job_token_scopes:
type: array
nullable: true
items:
type: string
description: "Scopes the token of this tool's jobs is restricted to, on top of the agent step's own restriction: a tool can only narrow it."
value:
$ref: '#/components/schemas/ToolValue'
required:
system_prompts/auto-generated/flow.md
+1 -1
View File
File diff suppressed because one or more lines are too long
system_prompts/auto-generated/prompts.ts
+1 -1
View File
File diff suppressed because one or more lines are too long
system_prompts/auto-generated/skills/write-flow/SKILL.md
+1 -1
View File
File diff suppressed because one or more lines are too long