fix: store per-job scopes on job_perms instead of v2_job, pin inline runs to the checked version

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-10-02 19:24:43 +02:00
co-authored by Claude Opus 5.5
parent d240e331a3
commit fc9fec6e0b
26 changed files with 152 additions and 405 deletions
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "WITH inserted_job AS (\n INSERT INTO v2_job (\n id, -- 1\n workspace_id, -- 2\n raw_code, -- 3\n raw_lock, -- 4\n raw_flow, -- 5\n tag, -- 6\n parent_job, -- 7\n created_by, -- 8\n permissioned_as, -- 9\n runnable_id, -- 10\n runnable_path, -- 11\n args, -- 12\n kind, -- 13\n trigger, -- 14\n script_lang, -- 15\n same_worker, -- 16\n pre_run_error, -- 17\n permissioned_as_email, -- 18\n visible_to_owner, -- 19\n flow_innermost_root_job, -- 20\n root_job, -- 38\n concurrent_limit, -- 21\n concurrency_time_window_s, -- 22\n timeout, -- 23\n flow_step_id, -- 24\n cache_ttl, -- 25\n priority, -- 26\n trigger_kind, -- 39\n script_entrypoint_override, -- 12\n preprocessed, -- 27,\n labels, -- 44\n job_token_scopes -- 47\n ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27,\n -- $44 (payload labels) merged with the labels of the runnable's folder, if any\n -- ($45/$46 are runnable_path/workspace_id again, kept separate to avoid parameter type conflicts)\n (SELECT CASE WHEN fl.labels IS NULL THEN $44\n ELSE (SELECT array_agg(DISTINCT lbl) FROM unnest(COALESCE($44, ARRAY[]::TEXT[]) || fl.labels) lbl)\n END\n FROM folder_labels($46, $45) AS fl(labels)), $47)\n ),\n inserted_runtime AS (\n INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)\n ),\n inserted_job_perms AS (\n INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, end_user_email)\n values ($1, $32, $33, $34, $35, $36, $37, $2, $41)\n ON CONFLICT (job_id) DO UPDATE SET email = EXCLUDED.email, username = EXCLUDED.username, is_admin = EXCLUDED.is_admin, is_operator = EXCLUDED.is_operator, folders = EXCLUDED.folders, groups = EXCLUDED.groups, workspace_id = EXCLUDED.workspace_id, end_user_email = EXCLUDED.end_user_email\n )\n INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority, cache_ignore_s3_path, runnable_settings_handle)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42, $43)",
"query": "WITH inserted_job AS (\n INSERT INTO v2_job (\n id, -- 1\n workspace_id, -- 2\n raw_code, -- 3\n raw_lock, -- 4\n raw_flow, -- 5\n tag, -- 6\n parent_job, -- 7\n created_by, -- 8\n permissioned_as, -- 9\n runnable_id, -- 10\n runnable_path, -- 11\n args, -- 12\n kind, -- 13\n trigger, -- 14\n script_lang, -- 15\n same_worker, -- 16\n pre_run_error, -- 17\n permissioned_as_email, -- 18\n visible_to_owner, -- 19\n flow_innermost_root_job, -- 20\n root_job, -- 38\n concurrent_limit, -- 21\n concurrency_time_window_s, -- 22\n timeout, -- 23\n flow_step_id, -- 24\n cache_ttl, -- 25\n priority, -- 26\n trigger_kind, -- 39\n script_entrypoint_override, -- 12\n preprocessed, -- 27,\n labels -- 44\n ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27,\n -- $44 (payload labels) merged with the labels of the runnable's folder, if any\n -- ($45/$46 are runnable_path/workspace_id again, kept separate to avoid parameter type conflicts)\n (SELECT CASE WHEN fl.labels IS NULL THEN $44\n ELSE (SELECT array_agg(DISTINCT lbl) FROM unnest(COALESCE($44, ARRAY[]::TEXT[]) || fl.labels) lbl)\n END\n FROM folder_labels($46, $45) AS fl(labels)))\n ),\n inserted_runtime AS (\n INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)\n ),\n inserted_job_perms AS (\n INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, end_user_email, job_token_scopes)\n values ($1, $32, $33, $34, $35, $36, $37, $2, $41, $47)\n ON CONFLICT (job_id) DO UPDATE SET email = EXCLUDED.email, username = EXCLUDED.username, is_admin = EXCLUDED.is_admin, is_operator = EXCLUDED.is_operator, folders = EXCLUDED.folders, groups = EXCLUDED.groups, workspace_id = EXCLUDED.workspace_id, end_user_email = EXCLUDED.end_user_email, job_token_scopes = EXCLUDED.job_token_scopes\n )\n INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority, cache_ignore_s3_path, runnable_settings_handle)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42, $43)",
"describe": {
"columns": [],
"parameters": {
@@ -148,5 +148,5 @@
},
"nullable": []
},
"hash": "921bdbe70e936935823fd202894a91a02d6bd218d21a9f43341cca846a5d276e"
"hash": "095fa75c60724664f0355cd6e3a64f84a0f1a31cd6732c0ab994a2d6ac3c3eec"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT job_token_scopes FROM v2_job WHERE id = $1",
"query": "SELECT job_token_scopes FROM job_perms WHERE job_id = $1",
"describe": {
"columns": [
{
@@ -18,5 +18,5 @@
true
]
},
"hash": "0312002d090d7233909922a96a756cb6311ca99b9a7343949ab653e404f65c23"
"hash": "13f358b7c76cda0b24fcb17dbbe10a00333bbf3b8b43bbccc7a6e1926d82478c"
}
@@ -1,139 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO v2_job (\n id, raw_code, raw_lock, raw_flow, tag, workspace_id, created_at, created_by,\n permissioned_as, permissioned_as_email, kind, runnable_id, runnable_path,\n parent_job, root_job, script_lang, script_entrypoint_override, flow_step,\n flow_step_id, flow_innermost_root_job, trigger, trigger_kind, same_worker,\n visible_to_owner, concurrent_limit, concurrency_time_window_s, cache_ttl,\n timeout, priority, preprocessed, args, labels, pre_run_error, job_token_scopes\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,\n $11, $12, $13, $14, $15, $16, $17, $18, $19, $20,\n $21, $22, $23, $24, $25, $26, $27, $28, $29, $30,\n $31, $32, $33, $34\n )\n ON CONFLICT (id) DO NOTHING\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Text",
"Text",
"Jsonb",
"Varchar",
"Varchar",
"Timestamptz",
"Varchar",
"Varchar",
"Varchar",
{
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
"aiagent",
"unassigned_script",
"unassigned_flow",
"unassigned_singlestepflow"
]
}
}
},
"Int8",
"Varchar",
"Uuid",
"Uuid",
{
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb",
"nu",
"java",
"duckdb",
"ruby",
"rlang",
"dbt"
]
}
}
},
"Varchar",
"Int4",
"Varchar",
"Uuid",
"Varchar",
{
"Custom": {
"name": "job_trigger_kind",
"kind": {
"Enum": [
"webhook",
"http",
"websocket",
"kafka",
"email",
"nats",
"schedule",
"app",
"ui",
"postgres",
"sqs",
"gcp",
"mqtt",
"nextcloud",
"google",
"ci_test",
"github",
"azure",
"asset",
"freshness",
"amqp"
]
}
}
},
"Bool",
"Bool",
"Int4",
"Int4",
"Int4",
"Int4",
"Int2",
"Bool",
"Jsonb",
"TextArray",
"Text",
"TextArray"
]
},
"nullable": []
},
"hash": "310bc5d8444925455ac2ff0904f91f36d84fb4797caccc11ea74bb7bc9890854"
}
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT job_token_scopes FROM v2_job WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "job_token_scopes",
"type_info": "TextArray"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true
]
},
"hash": "32f92624baf4eed5137af20edf8236428a28c455b75e8fa96d9512202cbd936d"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n v2_job_queue.runnable_settings_handle,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n cache_ignore_s3_path,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: TriggerKindLabel\",\n visible_to_owner,\n NULL as permissioned_as_end_user_email,\n job_token_scopes\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1",
"query": "SELECT\n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n v2_job_queue.runnable_settings_handle,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n cache_ignore_s3_path,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: TriggerKindLabel\",\n visible_to_owner,\n NULL as permissioned_as_end_user_email,\n job_perms.job_token_scopes\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id\n LEFT JOIN job_perms ON job_perms.job_id = v2_job_queue.id WHERE v2_job_queue.id = $1",
"describe": {
"columns": [
{
@@ -320,5 +320,5 @@
true
]
},
"hash": "eb1ea11d391e5860aa4a6e9d4c1ecac07c2e0998c2cee48f254915690d8270d0"
"hash": "333f7116d660756c36dfc0aa2b8d7ebd10a38cbe5724a554a12291c287b67f6d"
}
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, job_token_scopes)\n SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8, $9",
"describe": {
"columns": [],
"parameters": {
"Left": [
"UuidArray",
"Varchar",
"Varchar",
"Bool",
"Bool",
"JsonbArray",
"TextArray",
"Varchar",
"TextArray"
]
},
"nullable": []
},
"hash": "4cc81f5e04caefa4f170807c0ce96c69f22d9ea9dc3586099553a8f0e76aad82"
}
@@ -1,101 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH uuid_table as (\n select gen_random_uuid() as uuid from generate_series(1, $16)\n )\n INSERT INTO v2_job\n (id, workspace_id, raw_code, raw_lock, raw_flow, tag, runnable_id, runnable_path, kind,\n script_lang, created_by, permissioned_as, permissioned_as_email, concurrent_limit,\n concurrency_time_window_s, timeout, args, job_token_scopes)\n (SELECT uuid, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,\n ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $17 FROM uuid_table)\n RETURNING id AS \"id!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Jsonb",
"Varchar",
"Int8",
"Varchar",
{
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlestepflow",
"flowscript",
"flownode",
"appscript",
"aiagent",
"unassigned_script",
"unassigned_flow",
"unassigned_singlestepflow"
]
}
}
},
{
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb",
"nu",
"java",
"duckdb",
"ruby",
"rlang",
"dbt"
]
}
}
},
"Varchar",
"Varchar",
"Varchar",
"Int4",
"Int4",
"Int4",
"Int4",
"TextArray"
]
},
"nullable": [
false
]
},
"hash": "4ff18f19e633d8c649287097372fa7969335eccad727b9951ebe4bdc8eb08823"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE v2_job_status SET\n flow_status = JSONB_SET(\n JSONB_SET(v2_job_status.flow_status, $2::TEXT[], $3),\n $4::TEXT[], $5\n ) - $6::TEXT[],\n flow_leaf_jobs = CASE\n WHEN $7::TEXT IS NULL\n OR COALESCE(v2_job.flow_innermost_root_job, v2_job_status.id) <> v2_job_status.id\n THEN v2_job_status.flow_leaf_jobs\n ELSE JSONB_SET(COALESCE(v2_job_status.flow_leaf_jobs, '{}'::JSONB), ARRAY[$7::TEXT], $8) END\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id\n WHERE v2_job_status.id = $1 AND v2_job_queue.id = $1\n RETURNING\n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n v2_job_queue.runnable_settings_handle,\n v2_job_queue.scheduled_for,\n v2_job.runnable_path,\n v2_job.kind as \"kind: JobKind\",\n v2_job.runnable_id as \"runnable_id: ScriptHash\",\n v2_job_queue.canceled_reason,\n v2_job_queue.canceled_by,\n v2_job.permissioned_as,\n v2_job.permissioned_as_email,\n v2_job_status.flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n v2_job.script_lang as \"script_lang: ScriptLang\",\n v2_job.same_worker,\n v2_job.pre_run_error,\n v2_job.concurrent_limit,\n v2_job.concurrency_time_window_s,\n v2_job.flow_innermost_root_job,\n v2_job.root_job,\n v2_job.timeout,\n v2_job.flow_step_id,\n v2_job.cache_ttl,\n v2_job_queue.cache_ignore_s3_path,\n v2_job_queue.priority,\n v2_job.preprocessed,\n v2_job.script_entrypoint_override,\n v2_job.trigger,\n v2_job.trigger_kind as \"trigger_kind: TriggerKindLabel\",\n v2_job.visible_to_owner,\n NULL as permissioned_as_end_user_email,\n v2_job.job_token_scopes",
"query": "UPDATE v2_job_status SET\n flow_status = JSONB_SET(\n JSONB_SET(v2_job_status.flow_status, $2::TEXT[], $3),\n $4::TEXT[], $5\n ) - $6::TEXT[],\n flow_leaf_jobs = CASE\n WHEN $7::TEXT IS NULL\n OR COALESCE(v2_job.flow_innermost_root_job, v2_job_status.id) <> v2_job_status.id\n THEN v2_job_status.flow_leaf_jobs\n ELSE JSONB_SET(COALESCE(v2_job_status.flow_leaf_jobs, '{}'::JSONB), ARRAY[$7::TEXT], $8) END\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id\n LEFT JOIN job_perms ON job_perms.job_id = v2_job_queue.id\n WHERE v2_job_status.id = $1 AND v2_job_queue.id = $1\n RETURNING\n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n v2_job_queue.runnable_settings_handle,\n v2_job_queue.scheduled_for,\n v2_job.runnable_path,\n v2_job.kind as \"kind: JobKind\",\n v2_job.runnable_id as \"runnable_id: ScriptHash\",\n v2_job_queue.canceled_reason,\n v2_job_queue.canceled_by,\n v2_job.permissioned_as,\n v2_job.permissioned_as_email,\n v2_job_status.flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n v2_job.script_lang as \"script_lang: ScriptLang\",\n v2_job.same_worker,\n v2_job.pre_run_error,\n v2_job.concurrent_limit,\n v2_job.concurrency_time_window_s,\n v2_job.flow_innermost_root_job,\n v2_job.root_job,\n v2_job.timeout,\n v2_job.flow_step_id,\n v2_job.cache_ttl,\n v2_job_queue.cache_ignore_s3_path,\n v2_job_queue.priority,\n v2_job.preprocessed,\n v2_job.script_entrypoint_override,\n v2_job.trigger,\n v2_job.trigger_kind as \"trigger_kind: TriggerKindLabel\",\n v2_job.visible_to_owner,\n NULL as permissioned_as_end_user_email,\n job_perms.job_token_scopes",
"describe": {
"columns": [
{
@@ -327,5 +327,5 @@
true
]
},
"hash": "4d6dd480f3118dc5e099538811d3b9a1e996440f056b80f3a4e0ef3878e00198"
"hash": "6445041033c95fcf52c79c33f34cf6c41fb443f8bd91f12bce0e53f6e91bfb16"
}
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT job_token_scopes IS NOT NULL FROM script WHERE path = $1 AND workspace_id = $2 AND archived = false AND deleted = false ORDER BY created_at DESC LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "c6faa49fa9385412c8f5107864b24bb83ddb9bbb2dad215d8568ae64104bdcb7"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT p.email, p.username, p.is_admin, p.is_operator, p.groups, p.folders, p.end_user_email,\n j.job_token_scopes\n FROM job_perms p LEFT JOIN v2_job j ON j.id = p.job_id\n WHERE p.job_id = $1 AND p.workspace_id = $2",
"query": "SELECT email, username, is_admin, is_operator, groups, folders, end_user_email,\n job_token_scopes\n FROM job_perms WHERE job_id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
@@ -61,5 +61,5 @@
true
]
},
"hash": "ec72f5257c99202e43105043c70d70694ba2572f1b744fe69b191eb5b3d5a48f"
"hash": "c921b1a03e3181bbac351545941affc87853ff2d97f567a8817260cfef43e00e"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n v2_job.id,\n v2_job.raw_code,\n v2_job.raw_lock,\n v2_job.raw_flow as \"raw_flow: _\",\n v2_job.tag,\n v2_job.created_at,\n v2_job.created_by,\n v2_job.permissioned_as,\n v2_job.permissioned_as_email,\n v2_job.kind as \"kind: _\",\n v2_job.runnable_id,\n v2_job.runnable_path,\n v2_job.parent_job,\n v2_job.root_job,\n v2_job.script_lang as \"script_lang: _\",\n v2_job.script_entrypoint_override,\n v2_job.flow_step,\n v2_job.flow_step_id,\n v2_job.flow_innermost_root_job,\n v2_job.\"trigger\",\n v2_job.trigger_kind as \"trigger_kind: _\",\n v2_job.same_worker,\n v2_job.visible_to_owner,\n v2_job.concurrent_limit,\n v2_job.concurrency_time_window_s,\n v2_job.cache_ttl,\n v2_job.timeout,\n v2_job.priority,\n v2_job.preprocessed,\n v2_job.args as \"args: _\",\n v2_job.labels,\n v2_job.job_token_scopes,\n v2_job.pre_run_error,\n\n v2_job_queue.started_at,\n v2_job_queue.scheduled_for,\n v2_job_queue.running,\n v2_job_queue.canceled_by,\n v2_job_queue.canceled_reason,\n v2_job_queue.suspend,\n v2_job_queue.suspend_until,\n v2_job_queue.worker,\n v2_job_queue.extras as \"extras: _\",\n\n v2_job_runtime.ping,\n v2_job_runtime.memory_peak,\n\n v2_job_status.flow_status as \"flow_status: _\",\n v2_job_status.flow_leaf_jobs as \"flow_leaf_jobs: _\",\n v2_job_status.workflow_as_code_status as \"workflow_as_code_status: _\",\n\n concurrency_key.key as \"concurrency_key?\"\n FROM v2_job_queue\n INNER JOIN v2_job ON v2_job.id = v2_job_queue.id\n LEFT JOIN v2_job_runtime ON v2_job_runtime.id = v2_job_queue.id\n LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id\n LEFT JOIN concurrency_key ON concurrency_key.job_id = v2_job_queue.id\n WHERE v2_job_queue.workspace_id = $1\n AND v2_job_queue.running = false\n AND v2_job.parent_job IS NULL\n AND v2_job.trigger_kind IS DISTINCT FROM 'schedule'\n ORDER BY v2_job.created_at DESC\n LIMIT $2\n OFFSET $3\n ",
"query": "\n SELECT\n v2_job.id,\n v2_job.raw_code,\n v2_job.raw_lock,\n v2_job.raw_flow as \"raw_flow: _\",\n v2_job.tag,\n v2_job.created_at,\n v2_job.created_by,\n v2_job.permissioned_as,\n v2_job.permissioned_as_email,\n v2_job.kind as \"kind: _\",\n v2_job.runnable_id,\n v2_job.runnable_path,\n v2_job.parent_job,\n v2_job.root_job,\n v2_job.script_lang as \"script_lang: _\",\n v2_job.script_entrypoint_override,\n v2_job.flow_step,\n v2_job.flow_step_id,\n v2_job.flow_innermost_root_job,\n v2_job.\"trigger\",\n v2_job.trigger_kind as \"trigger_kind: _\",\n v2_job.same_worker,\n v2_job.visible_to_owner,\n v2_job.concurrent_limit,\n v2_job.concurrency_time_window_s,\n v2_job.cache_ttl,\n v2_job.timeout,\n v2_job.priority,\n v2_job.preprocessed,\n v2_job.args as \"args: _\",\n v2_job.labels,\n job_perms.job_token_scopes as \"job_token_scopes?\",\n v2_job.pre_run_error,\n\n v2_job_queue.started_at,\n v2_job_queue.scheduled_for,\n v2_job_queue.running,\n v2_job_queue.canceled_by,\n v2_job_queue.canceled_reason,\n v2_job_queue.suspend,\n v2_job_queue.suspend_until,\n v2_job_queue.worker,\n v2_job_queue.extras as \"extras: _\",\n\n v2_job_runtime.ping,\n v2_job_runtime.memory_peak,\n\n v2_job_status.flow_status as \"flow_status: _\",\n v2_job_status.flow_leaf_jobs as \"flow_leaf_jobs: _\",\n v2_job_status.workflow_as_code_status as \"workflow_as_code_status: _\",\n\n concurrency_key.key as \"concurrency_key?\"\n FROM v2_job_queue\n INNER JOIN v2_job ON v2_job.id = v2_job_queue.id\n LEFT JOIN job_perms ON job_perms.job_id = v2_job_queue.id\n LEFT JOIN v2_job_runtime ON v2_job_runtime.id = v2_job_queue.id\n LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id\n LEFT JOIN concurrency_key ON concurrency_key.job_id = v2_job_queue.id\n WHERE v2_job_queue.workspace_id = $1\n AND v2_job_queue.running = false\n AND v2_job.parent_job IS NULL\n AND v2_job.trigger_kind IS DISTINCT FROM 'schedule'\n ORDER BY v2_job.created_at DESC\n LIMIT $2\n OFFSET $3\n ",
"describe": {
"columns": [
{
@@ -252,7 +252,7 @@
},
{
"ordinal": 31,
"name": "job_token_scopes",
"name": "job_token_scopes?",
"type_info": "TextArray"
},
{
@@ -394,5 +394,5 @@
false
]
},
"hash": "984092e2bb9e4b380d7200524b4e10fdf4013033b128fbc47b65e240da847dc3"
"hash": "eefb19c05e6d9650b4dd1e50e10c26f01e117b75ab48e197c7d91d388865da74"
}
+1 -1
View File
@@ -1 +1 @@
56f2d36eff7edc4a2e36e89d8e3cb17588ceef38
9a8cd3db05b6c0ac69b05d97ae5355d98b812e0f
@@ -1,3 +1,3 @@
ALTER TABLE v2_job DROP COLUMN IF EXISTS job_token_scopes;
ALTER TABLE job_perms DROP COLUMN IF EXISTS job_token_scopes;
ALTER TABLE flow DROP COLUMN IF EXISTS job_token_scopes;
ALTER TABLE script DROP COLUMN IF EXISTS job_token_scopes;
@@ -1,3 +1,3 @@
ALTER TABLE script ADD COLUMN IF NOT EXISTS job_token_scopes text[];
ALTER TABLE flow ADD COLUMN IF NOT EXISTS job_token_scopes text[];
ALTER TABLE v2_job ADD COLUMN IF NOT EXISTS job_token_scopes text[];
ALTER TABLE job_perms ADD COLUMN IF NOT EXISTS job_token_scopes text[];
+2 -2
View File
@@ -132,7 +132,7 @@ input: id(uuid), workspace_id(char), runnable_id(char), runnable_type(runnable_t
FK: (workspace_id) -> workspace(id)
instance_group: name(char), summary(char), id(char), scim_display_name(char), external_id(char)
job_logs: job_id(uuid), workspace_id(char), created_at(ts), logs(text), log_offset(int), log_file_index(text[])
job_perms: job_id(uuid), email(char), username(char), is_admin(bool), is_operator(bool), created_at(ts), workspace_id(char), groups(text[]), folders(jsonb[]), end_user_email(char)
job_perms: job_id(uuid), email(char), username(char), is_admin(bool), is_operator(bool), created_at(ts), workspace_id(char), groups(text[]), folders(jsonb[]), end_user_email(char), job_token_scopes(text[])
job_resolution: job_id(uuid), workspace_id(char), resolved_at(ts), resolved_by(char), note(text), automatic(bool)
job_result_stream: job_id(uuid), workspace_id(text), stream(text)
job_result_stream_v2: job_id(uuid), workspace_id(text), stream(text), idx(int)
@@ -205,7 +205,7 @@ usr: workspace_id(char), username(char), email(char), is_admin(bool), created_at
FK: (workspace_id) -> workspace(id)
usr_to_group: workspace_id(char), group_(char), usr(char)
FK: (workspace_id, group_) -> group_(workspace_id, name) | (workspace_id) -> workspace(id)
v2_job: id(uuid), raw_code(text), raw_lock(text), raw_flow(jsonb), tag(char), workspace_id(char), created_at(ts), created_by(char), permissioned_as(char), permissioned_as_email(char), kind(job_kind), runnable_id(bigint), runnable_path(char), parent_job(uuid), root_job(uuid), script_lang(script_lang), script_entrypoint_override(char), flow_step(int), flow_step_id(char), flow_innermost_root_job(uuid), trigger(char), trigger_kind(job_trigger_kind), same_worker(bool), visible_to_owner(bool), concurrent_limit(int), concurrency_time_window_s(int), cache_ttl(int), timeout(int), priority(smallint), preprocessed(bool), args(jsonb), labels(text[]), pre_run_error(text), job_token_scopes(text[])
v2_job: id(uuid), raw_code(text), raw_lock(text), raw_flow(jsonb), tag(char), workspace_id(char), created_at(ts), created_by(char), permissioned_as(char), permissioned_as_email(char), kind(job_kind), runnable_id(bigint), runnable_path(char), parent_job(uuid), root_job(uuid), script_lang(script_lang), script_entrypoint_override(char), flow_step(int), flow_step_id(char), flow_innermost_root_job(uuid), trigger(char), trigger_kind(job_trigger_kind), same_worker(bool), visible_to_owner(bool), concurrent_limit(int), concurrency_time_window_s(int), cache_ttl(int), timeout(int), priority(smallint), preprocessed(bool), args(jsonb), labels(text[]), pre_run_error(text)
v2_job_completed: id(uuid), workspace_id(char), duration_ms(bigint), result(jsonb), deleted(bool), canceled_by(char), canceled_reason(text), flow_status(jsonb), started_at(ts), memory_peak(int), status(job_status), completed_at(ts), worker(char), workflow_as_code_status(jsonb), result_columns(text[]), retries(uuid[]), extras(jsonb)
v2_job_debounce_batch: id(uuid), debounce_batch(bigint)
v2_job_queue: id(uuid), workspace_id(char), created_at(ts), started_at(ts), scheduled_for(ts), running(bool), canceled_by(char), canceled_reason(text), suspend(int), suspend_until(ts), tag(char), priority(smallint), worker(char), extras(jsonb), cache_ignore_s3_path(bool), runnable_settings_handle(bigint)
+18 -5
View File
@@ -18,14 +18,24 @@ async fn insert_job(
parent: Option<&str>,
scopes: &[&str],
) -> anyhow::Result<()> {
let id = Uuid::parse_str(id)?;
sqlx::query(
"INSERT INTO v2_job (id, workspace_id, created_by, permissioned_as, permissioned_as_email,
kind, script_lang, runnable_path, tag, parent_job, job_token_scopes)
kind, script_lang, runnable_path, tag, parent_job)
VALUES ($1, 'test-workspace', 'test-user-3', 'u/test-user-3', 'test3@windmill.dev',
'script', 'deno', 'u/test-user-3/agent', 'deno', $2, $3)",
'script', 'deno', 'u/test-user-3/agent', 'deno', $2)",
)
.bind(Uuid::parse_str(id)?)
.bind(id)
.bind(parent.map(Uuid::parse_str).transpose()?)
.execute(db)
.await?;
sqlx::query(
"INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups,
workspace_id, job_token_scopes)
VALUES ($1, 'test3@windmill.dev', 'test-user-3', false, false, '{}', '{}',
'test-workspace', $2)",
)
.bind(id)
.bind(scopes)
.execute(db)
.await?;
@@ -153,7 +163,7 @@ async fn test_restricted_job_token_is_confined(db: Pool<Postgres>) -> anyhow::Re
assert_eq!(resp.status(), StatusCode::CREATED);
let child = Uuid::parse_str(&resp.text().await?)?;
let scopes: Option<Vec<String>> =
sqlx::query_scalar("SELECT job_token_scopes FROM v2_job WHERE id = $1")
sqlx::query_scalar("SELECT job_token_scopes FROM job_perms WHERE job_id = $1")
.bind(child)
.fetch_one(&db)
.await?;
@@ -196,7 +206,10 @@ async fn test_flow_steps_inherit_the_flow_restriction(db: Pool<Postgres>) -> any
.await;
let step: Option<Vec<String>> =
sqlx::query_scalar("SELECT job_token_scopes FROM v2_job WHERE parent_job = $1")
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",
)
.bind(flow.id)
.fetch_one(&db)
.await?;
+2 -2
View File
@@ -520,14 +520,14 @@ pub async fn caller_scope_ceiling(
return Ok(None);
};
let row = sqlx::query_scalar!(
"SELECT job_token_scopes FROM v2_job WHERE id = $1",
"SELECT job_token_scopes FROM job_perms WHERE job_id = $1",
job_id
)
.fetch_optional(db)
.await?;
Ok(match row {
Some(scopes) => scopes,
// The job row is gone (retention), so only the token's own claims are left.
// The job left the queue and its row was swept: only the token's own claims are left.
None => authed.scopes.clone(),
})
}
+14 -4
View File
@@ -278,7 +278,7 @@ pub async fn export_queued_jobs(
v2_job.preprocessed,
v2_job.args as "args: _",
v2_job.labels,
v2_job.job_token_scopes,
job_perms.job_token_scopes as "job_token_scopes?",
v2_job.pre_run_error,
v2_job_queue.started_at,
@@ -301,6 +301,7 @@ pub async fn export_queued_jobs(
concurrency_key.key as "concurrency_key?"
FROM v2_job_queue
INNER JOIN v2_job ON v2_job.id = v2_job_queue.id
LEFT JOIN job_perms ON job_perms.job_id = v2_job_queue.id
LEFT JOIN v2_job_runtime ON v2_job_runtime.id = v2_job_queue.id
LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id
LEFT JOIN concurrency_key ON concurrency_key.job_id = v2_job_queue.id
@@ -448,6 +449,16 @@ pub async fn import_queued_jobs(
));
}
// An imported job gets no `job_perms` row, so its token would be minted without the
// restriction it was queued with.
if let Some(job) = jobs.iter().find(|job| job.job_token_scopes.is_some()) {
return Err(error::Error::BadRequest(format!(
"Queued job {} restricts its job token and cannot be imported; run it again on \
this instance instead",
job.id
)));
}
let mut tx = user_db.begin(&authed).await?;
for job in jobs {
@@ -459,12 +470,12 @@ pub async fn import_queued_jobs(
parent_job, root_job, script_lang, script_entrypoint_override, flow_step,
flow_step_id, flow_innermost_root_job, trigger, trigger_kind, same_worker,
visible_to_owner, concurrent_limit, concurrency_time_window_s, cache_ttl,
timeout, priority, preprocessed, args, labels, pre_run_error, job_token_scopes
timeout, priority, preprocessed, args, labels, pre_run_error
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
$11, $12, $13, $14, $15, $16, $17, $18, $19, $20,
$21, $22, $23, $24, $25, $26, $27, $28, $29, $30,
$31, $32, $33, $34
$31, $32, $33
)
ON CONFLICT (id) DO NOTHING
"#,
@@ -501,7 +512,6 @@ pub async fn import_queued_jobs(
job.args as _,
job.labels as _,
job.pre_run_error,
job.job_token_scopes.as_deref() as Option<&[String]>,
)
.execute(&mut *tx)
.await?;
+32 -23
View File
@@ -8802,7 +8802,8 @@ async fn run_inline_script_by_path(
token,
db,
w_id,
InlineScriptTarget::Path(script_path.to_path().to_string()),
// Resolved to a version by `run_inline_script_inner`.
InlineScriptTarget::Path { path: script_path.to_path().to_string(), hash: 0 },
body.args,
Some(user_db),
)
@@ -8866,29 +8867,37 @@ async fn run_inline_script_inner(
user_db: Option<UserDB>,
) -> error::Result<Response> {
// An inline run executes with the caller's own token, so it cannot honour a script's
// `job_token_scopes`: such a script only runs as a job.
let restricted = match &target {
InlineScriptTarget::Path(path) => {
sqlx::query_scalar!(
"SELECT job_token_scopes IS NOT NULL FROM script WHERE path = $1 AND workspace_id = $2 \
AND archived = false AND deleted = false ORDER BY created_at DESC LIMIT 1",
path,
&w_id
)
.fetch_optional(&db)
.await?
// `job_token_scopes`: such a script only runs as a job. A path is resolved here, once, and
// the version checked is the version run.
let (target, restricted) = match target {
InlineScriptTarget::Path { path, .. } => {
let authed_ref = authed.to_authed_ref();
let info = get_latest_deployed_hash_for_path(
user_db
.as_ref()
.map(|db| UserDbWithAuthed { db: db.clone(), authed: &authed_ref }),
db.clone(),
&w_id,
&path,
)
.await?;
let restricted = info.job_token_scopes.is_some();
(InlineScriptTarget::Path { path, hash: info.hash }, restricted)
}
InlineScriptTarget::Hash(hash) => {
sqlx::query_scalar!(
"SELECT job_token_scopes IS NOT NULL FROM script WHERE hash = $1 AND workspace_id = $2",
hash,
&w_id
)
let restricted = sqlx::query_scalar!(
"SELECT job_token_scopes IS NOT NULL FROM script WHERE hash = $1 AND workspace_id = $2",
hash,
&w_id
)
.fetch_optional(&db)
.await?
.flatten()
.unwrap_or(false);
(InlineScriptTarget::Hash(hash), restricted)
}
};
if restricted.flatten().unwrap_or(false) {
if restricted {
return Err(Error::BadRequest(
"This script restricts its job token (job_token_scopes) and cannot be run inline; \
run it as a job instead"
@@ -9641,9 +9650,9 @@ async fn add_batch_jobs(
INSERT INTO v2_job
(id, workspace_id, raw_code, raw_lock, raw_flow, tag, runnable_id, runnable_path, kind,
script_lang, created_by, permissioned_as, permissioned_as_email, concurrent_limit,
concurrency_time_window_s, timeout, args, job_token_scopes)
concurrency_time_window_s, timeout, args)
(SELECT uuid, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,
('{ "uuid": "' || uuid || '" }')::jsonb, $17 FROM uuid_table)
('{ "uuid": "' || uuid || '" }')::jsonb FROM uuid_table)
RETURNING id AS "id!""#,
w_id,
raw_code,
@@ -9661,7 +9670,6 @@ async fn add_batch_jobs(
concurrency_time_window_s,
timeout,
n,
job_token_scopes.as_deref() as Option<&[String]>,
)
.fetch_all(&mut *tx)
.await?;
@@ -9690,8 +9698,8 @@ async fn add_batch_jobs(
.await?;
sqlx::query!(
"INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id)
SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8",
"INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, job_token_scopes)
SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8, $9",
&uuids,
authed.email,
authed.username,
@@ -9700,6 +9708,7 @@ async fn add_batch_jobs(
&[],
&[],
w_id,
job_token_scopes.as_deref() as Option<&[String]>,
)
.execute(&mut *tx)
.await?;
+8 -14
View File
@@ -362,7 +362,8 @@ pub struct JobPerms {
pub groups: Vec<String>,
pub folders: Vec<serde_json::Value>,
pub end_user_email: Option<String>,
/// The job's effective `v2_job.job_token_scopes`, which its token is minted with.
/// The job's effective scopes, stored on its `job_perms` row at push and minted into its
/// token.
#[serde(default)]
pub job_token_scopes: Option<Vec<String>>,
}
@@ -657,10 +658,9 @@ pub async fn get_job_perms<'a, E: sqlx::PgExecutor<'a>>(
) -> sqlx::Result<Option<JobPerms>> {
sqlx::query_as!(
JobPerms,
"SELECT p.email, p.username, p.is_admin, p.is_operator, p.groups, p.folders, p.end_user_email,
j.job_token_scopes
FROM job_perms p LEFT JOIN v2_job j ON j.id = p.job_id
WHERE p.job_id = $1 AND p.workspace_id = $2",
"SELECT email, username, is_admin, is_operator, groups, folders, end_user_email,
job_token_scopes
FROM job_perms WHERE job_id = $1 AND workspace_id = $2",
job_id,
w_id
)
@@ -756,6 +756,8 @@ pub async fn create_token_for_owner(
let scopes = jp.job_token_scopes.take();
(jp.into(), scopes)
}
// Push writes a job's `job_perms` row and its scopes in one statement, so a job with no
// row was never restricted.
_ => {
tracing::warn!("Could not get permissions for job {job_id} from job_perms table, getting permissions directly...");
let authed = fetch_authed_from_permissioned_as(owner, email, w_id, db)
@@ -765,15 +767,7 @@ pub async fn create_token_for_owner(
"Could not get permissions directly for job {job_id}: {e:#}"
))
})?;
let scopes = sqlx::query_scalar!(
"SELECT job_token_scopes FROM v2_job WHERE id = $1 AND workspace_id = $2",
job_id,
w_id
)
.fetch_optional(db)
.await?
.flatten();
(authed, scopes)
(authed, None)
}
};
+2 -1
View File
@@ -484,7 +484,8 @@ pub struct RunInlinePreviewScriptFnParams {
}
pub enum InlineScriptTarget {
Path(String),
/// A script addressed by path, pinned to the version the caller resolved and checked.
Path { path: String, hash: i64 },
Hash(i64),
}
+2 -2
View File
@@ -801,7 +801,7 @@ fn format_pull_query(peek: String) -> String {
flow_innermost_root_job, root_job, flow_step_id,
same_worker, pre_run_error, visible_to_owner, tag, concurrent_limit,
concurrency_time_window_s, timeout, cache_ttl, priority, raw_code, raw_lock,
raw_flow, script_entrypoint_override, preprocessed, job_token_scopes
raw_flow, script_entrypoint_override, preprocessed
FROM v2_job
WHERE id = (SELECT id FROM peek)
) SELECT j.id, j.workspace_id, j.parent_job, j.created_by, started_at, scheduled_for,
@@ -811,7 +811,7 @@ fn format_pull_query(peek: String) -> String {
j.same_worker, j.pre_run_error, j.visible_to_owner,
j.tag, j.concurrent_limit, j.concurrency_time_window_s, j.flow_innermost_root_job, j.root_job,
j.timeout, j.flow_step_id, j.cache_ttl, q.cache_ignore_s3_path, q.runnable_settings_handle, j.priority, j.raw_code, j.raw_lock, j.raw_flow,
j.script_entrypoint_override, j.preprocessed, j.job_token_scopes, COALESCE(pj.runnable_path, j.args->>'_FLOW_PATH') as parent_runnable_path,
j.script_entrypoint_override, j.preprocessed, p.job_token_scopes, COALESCE(pj.runnable_path, j.args->>'_FLOW_PATH') as parent_runnable_path,
COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin,
p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email
FROM q, j
+26 -22
View File
@@ -769,10 +769,11 @@ async fn restart_perpetual_runs_at_path(
/// The effective token scopes of an earlier job, as the ceiling of a job that re-runs it
/// (a retry, a perpetual restart): the re-run never holds a wider token than the run it
/// replaces.
/// replaces. Read while that job still holds its `job_perms` row, which the monitor sweeps
/// once the job has left the queue.
async fn stored_job_token_scopes(db: &DB, job_id: Uuid) -> Result<Option<Vec<String>>, Error> {
Ok(sqlx::query_scalar!(
"SELECT job_token_scopes FROM v2_job WHERE id = $1",
"SELECT job_token_scopes FROM job_perms WHERE job_id = $1",
job_id
)
.fetch_optional(db)
@@ -3597,7 +3598,7 @@ pub struct MiniPulledJob {
pub visible_to_owner: bool,
pub permissioned_as_end_user_email: Option<String>,
pub runnable_settings_handle: Option<i64>,
/// The job's effective token scopes (`v2_job.job_token_scopes`), minted into its token.
/// The job's effective token scopes (`job_perms.job_token_scopes`), minted into its token.
/// No `sqlx(default)`: a query that forgets the column must fail rather than mint an
/// unrestricted token.
#[serde(default)]
@@ -4066,8 +4067,9 @@ pub async fn get_mini_pulled_job<'c>(
trigger_kind as \"trigger_kind: TriggerKindLabel\",
visible_to_owner,
NULL as permissioned_as_end_user_email,
job_token_scopes
FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1",
job_perms.job_token_scopes
FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id
LEFT JOIN job_perms ON job_perms.job_id = v2_job_queue.id WHERE v2_job_queue.id = $1",
job_id,
)
.fetch_optional(e)
@@ -7040,16 +7042,19 @@ async fn push_inner<'c, 'd>(
memory_id: None,
no_inherited_flow_env: false,
};
// The restart runs the completed job's stored definition, so it keeps that job's
// restriction rather than reading the flow's current setting.
let job_token_scopes = sqlx::query_scalar::<_, Option<Vec<String>>>(
"SELECT job_token_scopes FROM v2_job WHERE id = $1 AND workspace_id = $2",
)
.bind(completed_job_id)
.bind(workspace_id)
.fetch_optional(db)
.await?
.flatten();
// The completed job's own scopes are gone with its `job_perms` row, so the restart
// takes the flow's current setting (and the restarting caller's ceiling).
let job_token_scopes = match &flow_path {
Some(flow_path) => sqlx::query_scalar::<_, Option<Vec<String>>>(
"SELECT job_token_scopes FROM flow WHERE path = $1 AND workspace_id = $2",
)
.bind(flow_path)
.bind(workspace_id)
.fetch_optional(db)
.await?
.flatten(),
None => None,
};
let value = flow_data.value();
let priority = value.priority;
let concurrency_settings = value.concurrency_settings.clone();
@@ -7508,8 +7513,7 @@ async fn push_inner<'c, 'd>(
trigger_kind, -- 39
script_entrypoint_override, -- 12
preprocessed, -- 27,
labels, -- 44
job_token_scopes -- 47
labels -- 44
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
$19, $20, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind,
($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27,
@@ -7518,15 +7522,15 @@ async fn push_inner<'c, 'd>(
(SELECT CASE WHEN fl.labels IS NULL THEN $44
ELSE (SELECT array_agg(DISTINCT lbl) FROM unnest(COALESCE($44, ARRAY[]::TEXT[]) || fl.labels) lbl)
END
FROM folder_labels($46, $45) AS fl(labels)), $47)
FROM folder_labels($46, $45) AS fl(labels)))
),
inserted_runtime AS (
INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)
),
inserted_job_perms AS (
INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, end_user_email)
values ($1, $32, $33, $34, $35, $36, $37, $2, $41)
ON CONFLICT (job_id) DO UPDATE SET email = EXCLUDED.email, username = EXCLUDED.username, is_admin = EXCLUDED.is_admin, is_operator = EXCLUDED.is_operator, folders = EXCLUDED.folders, groups = EXCLUDED.groups, workspace_id = EXCLUDED.workspace_id, end_user_email = EXCLUDED.end_user_email
INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, end_user_email, job_token_scopes)
values ($1, $32, $33, $34, $35, $36, $37, $2, $41, $47)
ON CONFLICT (job_id) DO UPDATE SET email = EXCLUDED.email, username = EXCLUDED.username, is_admin = EXCLUDED.is_admin, is_operator = EXCLUDED.is_operator, folders = EXCLUDED.folders, groups = EXCLUDED.groups, workspace_id = EXCLUDED.workspace_id, end_user_email = EXCLUDED.end_user_email, job_token_scopes = EXCLUDED.job_token_scopes
)
INSERT INTO v2_job_queue
(workspace_id, id, running, scheduled_for, started_at, tag, priority, cache_ignore_s3_path, runnable_settings_handle)
@@ -8545,7 +8549,7 @@ pub async fn get_same_worker_job(
v2_job.trigger,
v2_job.trigger_kind,
v2_job.visible_to_owner,
v2_job.job_token_scopes,
p.job_token_scopes,
v2_job.raw_code,
v2_job.raw_lock,
v2_job.raw_flow,
+2 -24
View File
@@ -7240,30 +7240,8 @@ pub fn init_worker_internal_server_inline_utils(
run_inline_script: Arc::new(|params: RunInlineScriptFnParams| {
Box::pin(async move {
let (script_hash, runnable_path) = match params.target {
InlineScriptTarget::Path(ref path) => {
let db = params
.conn
.as_sql()
.ok_or_else(|| {
error::Error::InternalErr(
"run_inline_script by path requires a SQL connection"
.to_string(),
)
})?
.clone();
let authed_ref = params.user_db.as_ref().map(|(_, a)| a.to_authed_ref());
let user_db_authed =
params.user_db.as_ref().zip(authed_ref.as_ref()).map(
|((udb, _), ar)| UserDbWithAuthed { db: udb.clone(), authed: ar },
);
let script_hash_info = get_latest_deployed_hash_for_path(
user_db_authed,
db,
&params.workspace_id,
path,
)
.await?;
(ScriptHash(script_hash_info.hash), Some(path.clone()))
InlineScriptTarget::Path { ref path, hash } => {
(ScriptHash(hash), Some(path.clone()))
}
InlineScriptTarget::Hash(hash) => (ScriptHash(hash), None),
};
+2 -1
View File
@@ -2366,6 +2366,7 @@ async fn advance_flow_status(
THEN v2_job_status.flow_leaf_jobs
ELSE JSONB_SET(COALESCE(v2_job_status.flow_leaf_jobs, '{}'::JSONB), ARRAY[$7::TEXT], $8) END
FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id
LEFT JOIN job_perms ON job_perms.job_id = v2_job_queue.id
WHERE v2_job_status.id = $1 AND v2_job_queue.id = $1
RETURNING
v2_job_queue.workspace_id,
@@ -2403,7 +2404,7 @@ async fn advance_flow_status(
v2_job.trigger_kind as \"trigger_kind: TriggerKindLabel\",
v2_job.visible_to_owner,
NULL as permissioned_as_end_user_email,
v2_job.job_token_scopes",
job_perms.job_token_scopes",
flow,
&step_path as &[&str],
step,
+4 -3
View File
@@ -53,12 +53,13 @@ Symbols, not line numbers, are cited: they drift less.
for every reader until turned off. The resource ACL is the only control, so granting write on a
folder grants that influence.
- **Restricted job tokens** (`job_token_scopes` on scripts and flows): a job's effective scopes
are stored in `v2_job.job_token_scopes` at push and minted into its token
are stored on its `job_perms` row at push and minted into its token
(`create_token_for_owner`); an empty set is minted as `NO_API_ACCESS_SCOPE`, since
`scopes: Some([])` reads as unscoped. Every `push` takes a `scope_ceiling`, and the job gets the
ceiling ∩ the target's own setting: flow steps take their flow job's, agent tools their agent's,
WAC children their parent's, retries and restarts the run they replace, and API pushes the
calling job's (`caller_scope_ceiling`) — a scoped user or webhook token caps nothing. Schedules,
WAC children their parent's, retries the run they replace, and API pushes the calling job's
(`caller_scope_ceiling`) — a scoped user or webhook token caps nothing. The row is swept once a
job leaves the queue, so a restart of a completed flow takes the flow's current setting. Schedules,
triggers and error/success handlers start from the target's own setting. A restricted token
keeps only the runtime routes about its own job (`is_own_job_runtime_route`) and the reads of
its own flow run (`flow_run_read_route_job`, checked against its lineage: the orchestrator