From 16eae212cc7267fa852731150917d5780e12e384 Mon Sep 17 00:00:00 2001 From: tristantr Date: Mon, 1 Jun 2026 13:42:45 +0200 Subject: [PATCH] fix(deploy-to-hub): gate hub publish endpoints + harden trigger detection - Add ApiAuthed + require_admin to all hub publish handlers; previously any workspace-authenticated session could trigger Hub-side writes attributed to the URL workspace via the shared HUB_DEV_TOKEN. - Track per-kind trigger fetch failures (triggerLoadErrors) so an EE-gated or transiently failing trigger service no longer silently maps to "0 triggers"; UI surfaces an amber badge listing the missing kinds and a toast warns the operator before publish. - Add workspaceLoadSeq cancellation so the parallel loadWorkspace + loadTriggers stop bleeding stale data when the workspace switches mid load. - Drop the silent effectiveSlug fallback to sanitizeSlug(hubName) when the Hub response can't be parsed; abort the publish instead so items don't land under a slug the Hub never locked. Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/windmill-api/src/hub_publish.rs | 25 +- .../workspaceSettings/DeployToHub.svelte | 517 ++++++++++++++++-- .../workspaceSettings/projectBundle.ts | 26 +- 3 files changed, 509 insertions(+), 59 deletions(-) diff --git a/backend/windmill-api/src/hub_publish.rs b/backend/windmill-api/src/hub_publish.rs index 9669ba1e2b..9ff57d7667 100644 --- a/backend/windmill-api/src/hub_publish.rs +++ b/backend/windmill-api/src/hub_publish.rs @@ -1,3 +1,4 @@ +use crate::db::ApiAuthed; use crate::HTTP_CLIENT; use axum::{ extract::{Json, Path}, @@ -7,7 +8,7 @@ use axum::{ Router, }; use serde::{Deserialize, Serialize}; -use windmill_common::{error::Error, HUB_BASE_URL}; +use windmill_common::{error::Error, utils::require_admin, HUB_BASE_URL}; pub fn workspaced_service() -> Router { Router::new() @@ -36,6 +37,7 @@ struct PublishDraftBody { #[derive(Serialize)] struct HubWorkspaceBody { + source_id: String, slug: String, name: String, summary: String, @@ -44,12 +46,15 @@ struct HubWorkspaceBody { } async fn publish_draft( - Path(_workspace): Path, + authed: ApiAuthed, + Path(workspace): Path, Json(body): Json, ) -> Result { + require_admin(authed.is_admin, &authed.username)?; forward_to_hub( "/workspaces", &HubWorkspaceBody { + source_id: workspace, slug: body.slug, name: body.name, summary: body.summary, @@ -77,9 +82,11 @@ struct PublishScriptBody { } async fn publish_script( + authed: ApiAuthed, Path(_workspace): Path, Json(body): Json, ) -> Result { + require_admin(authed.is_admin, &authed.username)?; forward_to_hub("/scripts/add", &body).await } @@ -103,9 +110,11 @@ struct PublishFlowBody { } async fn publish_flow( + authed: ApiAuthed, Path(_workspace): Path, Json(body): Json, ) -> Result { + require_admin(authed.is_admin, &authed.username)?; forward_to_hub("/flows", &body).await } @@ -120,9 +129,11 @@ struct PublishAppBody { } async fn publish_app( + authed: ApiAuthed, Path(_workspace): Path, Json(body): Json, ) -> Result { + require_admin(authed.is_admin, &authed.username)?; forward_to_hub("/apps", &body).await } @@ -139,9 +150,11 @@ struct PublishRawAppBody { } async fn publish_raw_app( + authed: ApiAuthed, Path(_workspace): Path, Json(body): Json, ) -> Result { + require_admin(authed.is_admin, &authed.username)?; forward_to_hub("/raw_apps", &body).await } @@ -153,16 +166,20 @@ struct RecordingBody { } async fn publish_script_recording( + authed: ApiAuthed, Path((_workspace, ask_id)): Path<(String, i64)>, Json(body): Json, ) -> Result { + require_admin(authed.is_admin, &authed.username)?; forward_to_hub(&format!("/scripts/{}/recording", ask_id), &body).await } async fn publish_flow_recording( + authed: ApiAuthed, Path((_workspace, flow_id)): Path<(String, i64)>, Json(body): Json, ) -> Result { + require_admin(authed.is_admin, &authed.username)?; forward_to_hub(&format!("/flows/{}/recording", flow_id), &body).await } @@ -179,9 +196,11 @@ struct PublishResourceTypeBody { } async fn publish_resource_type( + authed: ApiAuthed, Path(_workspace): Path, Json(body): Json, ) -> Result { + require_admin(authed.is_admin, &authed.username)?; forward_to_hub( &format!("/workspaces/{}/resource_types", body.workspace_slug), &body, @@ -204,9 +223,11 @@ struct PublishResourcesBody { } async fn publish_resources( + authed: ApiAuthed, Path(_workspace): Path, Json(body): Json, ) -> Result { + require_admin(authed.is_admin, &authed.username)?; forward_to_hub( &format!("/workspaces/{}/resources", body.workspace_slug), &body, diff --git a/frontend/src/lib/components/workspaceSettings/DeployToHub.svelte b/frontend/src/lib/components/workspaceSettings/DeployToHub.svelte index 5a243800a6..5643a4411b 100644 --- a/frontend/src/lib/components/workspaceSettings/DeployToHub.svelte +++ b/frontend/src/lib/components/workspaceSettings/DeployToHub.svelte @@ -22,7 +22,18 @@ RawAppService, ResourceService, ScriptService, - WorkspaceService + WorkspaceService, + HttpTriggerService, + WebsocketTriggerService, + KafkaTriggerService, + NatsTriggerService, + SqsTriggerService, + MqttTriggerService, + GcpTriggerService, + AzureTriggerService, + PostgresTriggerService, + EmailTriggerService, + ScheduleService } from '$lib/gen' import { workspaceStore } from '$lib/stores' import { computeSecretUrl } from '$lib/components/apps/editor/appDeploy.svelte' @@ -41,17 +52,20 @@ import { Check, Cloud, + Code2, Copy, ExternalLink, GitCompare, Globe, Info, + LayoutDashboard, Loader2, Play, RotateCcw, TriangleAlert, X } from 'lucide-svelte' + import BarsStaggered from '$lib/components/icons/BarsStaggered.svelte' import Popover from '$lib/components/Popover.svelte' import type { Kind } from '$lib/utils_deployable' @@ -71,11 +85,65 @@ const canRecord = (k: Kind) => k === 'script' || k === 'flow' const canPublishApp = (k: Kind) => k === 'app' || k === 'raw_app' + type TriggerKindLabel = + | 'http' + | 'websocket' + | 'schedule' + | 'kafka' + | 'nats' + | 'sqs' + | 'mqtt' + | 'gcp' + | 'azure' + | 'postgres' + | 'email' + interface WorkspaceTrigger { + kind: TriggerKindLabel + path: string + script_path: string + is_flow: boolean + summary?: string + config: Record + } + const TRIGGER_KIND_BADGE: Record = { + http: 'HTTP', + websocket: 'WebSocket', + schedule: 'Schedule', + kafka: 'Kafka', + nats: 'NATS', + sqs: 'SQS', + mqtt: 'MQTT', + gcp: 'GCP Pub/Sub', + azure: 'Azure', + postgres: 'Postgres', + email: 'Email' + } + const TRIGGER_KIND_NOTE: Partial> = { + http: 'Webhook URL regenerates on import — re-register with the external service.', + websocket: 'Reconnect WebSocket auth after import if external service requires it.', + email: 'Email address regenerates on import.', + kafka: 'Verify Kafka broker access from the importing instance.', + nats: 'Verify NATS connection from the importing instance.', + gcp: 'Re-link GCP Pub/Sub subscription after import.', + azure: 'Re-link Azure Event Grid subscription after import.' + } + let phase = $state('predeploy') // Live workspace items (refreshed from API). Used in predeploy phase only. let workspaceItems = $state([]) // Snapshot frozen at deploy time. Used in draft / under_review / live phases. let draftItems = $state([]) + let workspaceTriggers = $state([]) + let triggersLoading = $state(false) + let triggerLoadErrors = $state>(new Set()) + let workspaceLoadSeq = 0 + let schedulePreviews = $state>({}) + const schedulePreviewsInFlight = new Set() + const dateFmt = new Intl.DateTimeFormat(undefined, { + dateStyle: 'medium', + timeStyle: 'short' + }) + let selectedFolders = $state([]) let availableFolders = $derived( Array.from( @@ -160,6 +228,7 @@ let publishing = $state(false) let resourceDrawer = $state() + let triggerDrawer = $state() let bundleDrawer = $state() let bundleName = $state('') let bundleSummary = $state('') @@ -169,6 +238,8 @@ let hubSummary = $state('') let hubReadme = $state('') + let effectiveSlug = $state('') + const delay = (ms: number) => new Promise((r) => setTimeout(r, ms)) function patchItem(key: string, patch: Partial) { @@ -189,7 +260,7 @@ return out } - async function loadWorkspace(workspace: string) { + async function loadWorkspace(workspace: string, seq: number) { loading = true try { const [apps, rawApps, flows, scripts, settings] = await Promise.all([ @@ -199,6 +270,7 @@ listAllPages((p) => ScriptService.listScripts({ workspace, ...p })), WorkspaceService.getSettings({ workspace }).catch(() => undefined) ]) + if (seq !== workspaceLoadSeq) return workspaceRateLimit = settings?.public_app_execution_limit_per_minute @@ -250,11 +322,81 @@ // Resources are NOT selectable items — they're dependencies derived from // the $res: references in the selected scripts/flows/apps and shown // read-only (see detectedResources). + if (seq !== workspaceLoadSeq) return workspaceItems = next } catch (e: any) { - sendUserToast(`Failed to load workspace items: ${e?.message ?? e}`, true) + if (seq === workspaceLoadSeq) { + sendUserToast(`Failed to load workspace items: ${e?.message ?? e}`, true) + } } finally { - loading = false + if (seq === workspaceLoadSeq) loading = false + } + } + + async function loadTriggers(workspace: string, seq: number) { + triggersLoading = true + const errors = new Set() + try { + const listOrMark = async (kind: TriggerKindLabel, p: Promise): Promise => { + try { + return await p + } catch { + errors.add(kind) + return [] + } + } + const [http, websocket, schedule, kafka, nats, sqs, mqtt, gcp, azure, postgres, email] = + await Promise.all([ + listOrMark('http', HttpTriggerService.listHttpTriggers({ workspace })), + listOrMark('websocket', WebsocketTriggerService.listWebsocketTriggers({ workspace })), + listOrMark('schedule', ScheduleService.listSchedules({ workspace })), + listOrMark('kafka', KafkaTriggerService.listKafkaTriggers({ workspace })), + listOrMark('nats', NatsTriggerService.listNatsTriggers({ workspace })), + listOrMark('sqs', SqsTriggerService.listSqsTriggers({ workspace })), + listOrMark('mqtt', MqttTriggerService.listMqttTriggers({ workspace })), + listOrMark('gcp', GcpTriggerService.listGcpTriggers({ workspace })), + listOrMark('azure', AzureTriggerService.listAzureTriggers({ workspace })), + listOrMark('postgres', PostgresTriggerService.listPostgresTriggers({ workspace })), + listOrMark('email', EmailTriggerService.listEmailTriggers({ workspace })) + ]) + if (seq !== workspaceLoadSeq) return + const normalize = ( + kind: TriggerKindLabel, + rows: Array> + ): WorkspaceTrigger[] => + rows.map((r) => ({ + kind, + path: r.path, + script_path: r.script_path, + is_flow: r.is_flow ?? false, + summary: r.summary, + config: r + })) + workspaceTriggers = [ + ...normalize('http', http), + ...normalize('websocket', websocket), + ...normalize('schedule', schedule), + ...normalize('kafka', kafka), + ...normalize('nats', nats), + ...normalize('sqs', sqs), + ...normalize('mqtt', mqtt), + ...normalize('gcp', gcp), + ...normalize('azure', azure), + ...normalize('postgres', postgres), + ...normalize('email', email) + ] + triggerLoadErrors = errors + if (errors.size > 0) { + const labels = Array.from(errors) + .map((k) => TRIGGER_KIND_BADGE[k]) + .join(', ') + sendUserToast( + `Could not list triggers for: ${labels}. The bundle may miss stubs from those kinds.`, + true + ) + } + } finally { + if (seq === workspaceLoadSeq) triggersLoading = false } } @@ -269,22 +411,148 @@ $effect(() => { if ($workspaceStore) { - loadWorkspace($workspaceStore) + const seq = ++workspaceLoadSeq + // Drop stale workspace state immediately so prior data can't bleed + // into the new workspace before the parallel fetches resolve. + workspaceItems = [] + workspaceTriggers = [] + triggerLoadErrors = new Set() + schedulePreviews = {} + loadWorkspace($workspaceStore, seq) + loadTriggers($workspaceStore, seq) } }) + let relevantTriggers = $derived.by(() => { + const selectedScripts = new Set( + selectedItems.filter((i) => i.kind === 'script').map((i) => i.path) + ) + const selectedFlows = new Set(selectedItems.filter((i) => i.kind === 'flow').map((i) => i.path)) + return workspaceTriggers.filter((t) => + t.is_flow ? selectedFlows.has(t.script_path) : selectedScripts.has(t.script_path) + ) + }) + + let triggersByKind = $derived.by(() => { + const out = new Map() + for (const t of relevantTriggers) { + const arr = out.get(t.kind) ?? [] + arr.push(t) + out.set(t.kind, arr) + } + return Array.from(out.entries()).sort((a, b) => a[0].localeCompare(b[0])) + }) + + $effect(() => { + for (const t of relevantTriggers) { + if (t.kind !== 'schedule') continue + const c = t.config as any + const key = `${c.schedule}|${c.timezone}` + if (schedulePreviews[key] || schedulePreviewsInFlight.has(key)) continue + schedulePreviewsInFlight.add(key) + ScheduleService.previewSchedule({ + requestBody: { + schedule: c.schedule, + timezone: c.timezone, + cron_version: c.cron_version ?? 'v2' + } + }) + .then((dates) => { + schedulePreviews = { ...schedulePreviews, [key]: dates.slice(0, 3) } + }) + .catch(() => {}) + .finally(() => schedulePreviewsInFlight.delete(key)) + } + }) + + function triggerDetails(t: WorkspaceTrigger): Array<{ label: string; value: string }> { + const c = t.config as any + const out: Array<{ label: string; value: string }> = [] + const push = (label: string, v: any) => { + if (v != null && v !== '') out.push({ label, value: String(v) }) + } + switch (t.kind) { + case 'http': + push('Route', `${(c.http_method ?? '').toUpperCase()} /${c.route_path ?? ''}`) + push('Auth', c.authentication_method) + break + case 'schedule': + push('Cron', c.schedule) + push('Timezone', c.timezone) + break + case 'websocket': + push('URL', c.url) + break + case 'kafka': + push('Resource', c.kafka_resource_path) + push('Group', c.group_id) + push('Topics', (c.topics ?? []).join(', ')) + break + case 'nats': + push('Resource', c.nats_resource_path) + push('Subjects', (c.subjects ?? []).join(', ')) + push('Jetstream', c.use_jetstream) + break + case 'sqs': + push('Queue', c.queue_url) + push('Resource', c.aws_resource_path) + break + case 'mqtt': + push('Resource', c.mqtt_resource_path) + push('Topics', (c.subscribe_topics ?? []).map((x: any) => x?.topic ?? x).join(', ')) + break + case 'gcp': + push('Resource', c.gcp_resource_path) + push('Topic', c.topic_id) + push('Subscription', c.subscription_id) + break + case 'azure': + push('Resource', c.azure_resource_path) + push('Scope', c.scope_resource_id) + push('Subscription', c.subscription_name) + break + case 'postgres': + push('Resource', c.postgres_resource_path) + push('Publication', c.publication_name) + break + case 'email': + push('Local part', c.local_part) + break + } + return out + } + + let runnableSummaryByPath = $derived.by(() => { + const m = new Map() + for (const it of workspaceItems) { + if (it.kind === 'script' || it.kind === 'flow') { + m.set(`${it.kind}:${it.path}`, it.summary) + } + } + return m + }) + function openBundle() { bundleName = hubName bundleSummary = hubSummary bundleReadme = hubReadme bundleDrawer?.openDrawer() } + // Mirror the Hub's SLUG_RE: ^[a-z0-9][a-z0-9-]{1,48}[a-z0-9]$ + // (3-50 chars, alphanumeric start/end, dashes only in the middle). function sanitizeSlug(s: string): string { return s .toLowerCase() .replace(/[_\s]+/g, '-') .replace(/[^a-z0-9-]/g, '') + .replace(/-+/g, '-') + .replace(/^-+|-+$/g, '') .slice(0, 50) + .replace(/-+$/g, '') + } + const SLUG_RE = /^[a-z0-9][a-z0-9-]{1,48}[a-z0-9]$/ + function isValidSlug(s: string): boolean { + return SLUG_RE.test(s) } async function confirmBundle() { @@ -299,17 +567,34 @@ headers: { 'Content-Type': 'application/json' }, credentials: 'include', body: JSON.stringify({ - slug: sanitizeSlug(hubSlug), + slug: effectiveSlug || sanitizeSlug(hubName), name: hubName, summary: hubSummary || hubName, readme: hubReadme || undefined }) }) + const text = await res.text() if (!res.ok) { - const text = await res.text() sendUserToast(`Hub draft creation failed: ${text}`, true) return } + // The Hub returns the effective slug (locked once a project exists). If the + // response can't be parsed or omits the slug, abort instead of guessing — + // proceeding with a client-computed slug risks landing items under a + // folder the Hub never created (proxy-rewritten body, server-side + // collision suffix, etc.). + let returnedSlug: string | undefined + try { + const parsed = JSON.parse(text) + if (typeof parsed?.slug === 'string') returnedSlug = parsed.slug + } catch { + // fallthrough — handled below + } + if (!returnedSlug) { + sendUserToast(`Hub did not return a slug. Aborting publish to avoid path drift.`, true) + return + } + effectiveSlug = returnedSlug } catch (e: any) { sendUserToast(`Hub draft creation failed: ${e?.message ?? e}`, true) return @@ -529,7 +814,7 @@ } let cancelled = false detectingResources = true - const slug = sanitizeSlug(hubSlug) + const slug = effectiveSlug || sanitizeSlug(hubName) const seed: ItemRef[] = selectedItems .filter((i) => i.kind !== 'resource') .map((i) => ({ kind: i.kind as ItemRef['kind'], path: i.path })) @@ -584,7 +869,8 @@ async function deployAll() { const workspace = $workspaceStore if (!workspace) return - const slug = sanitizeSlug(hubSlug) + // Use the Hub's effective (locked) slug so relocated paths match the project. + const slug = effectiveSlug || sanitizeSlug(hubName) deploying = true let failures = 0 try { @@ -695,7 +981,7 @@ // MOCK: would diff the live workspace against the submitted bundle and open a viewer. const workspace = $workspaceStore if (!workspace) return - await loadWorkspace(workspace) + await loadWorkspace(workspace, workspaceLoadSeq) const draftKeys = new Set(draftItems.map((i) => i.key)) const workspaceKeys = new Set(workspaceItems.map((i) => i.key)) const added = workspaceItems.filter((i) => !draftKeys.has(i.key)).length @@ -710,7 +996,7 @@ if (!workspace) return syncing = true try { - await loadWorkspace(workspace) + await loadWorkspace(workspace, workspaceLoadSeq) if (phase === 'draft') { const prev = new Map(draftItems.map((i) => [i.key, { rec: i.rec }])) draftItems = workspaceItems @@ -912,7 +1198,7 @@ const path = it.kind === 'script' ? 'scripts' : 'flows' await postHub(workspace, `/hub/${path}/${hubId}/recording`, { recording, - workspace_slug: sanitizeSlug(hubSlug) + workspace_slug: effectiveSlug || sanitizeSlug(hubName) }) recordings = { ...recordings, [it.key]: runJobId } patchItem(it.key, { rec: 'recorded' }) @@ -1071,13 +1357,13 @@ {/if} {#if phase === 'predeploy'} -
- +
+ Resource dependencies {#if detectingResources} {:else} - ({dependencyTypes.length}) + ({dependencyTypes.length}) {/if} Resource types the selected items depend on (whether passed as inputs or referenced @@ -1090,26 +1376,66 @@ No resource references detected in the current selection. {:else} -
- {#each dependencyTypes as r (r.resource_type)} - - {r.resource_type} - - {/each} - -
+ {r.resource_type} + + {/each} + + {/if} +
+ {/if} + {#if phase === 'predeploy'} +
+ + Triggers + {#if triggersLoading} + + {:else} + ({relevantTriggers.length}) + {/if} + + {#if triggerLoadErrors.size > 0} + + + Could not list: + {Array.from(triggerLoadErrors) + .map((k) => TRIGGER_KIND_BADGE[k]) + .join(', ')} + + {/if} + {#if relevantTriggers.length === 0} + No triggers reference the selected items. + {:else} + {#each triggersByKind as [kind, triggers] (kind)} + + {TRIGGER_KIND_BADGE[kind]} + ×{triggers.length} + + {/each} + {/if}
{/if} @@ -1506,7 +1832,13 @@ {#each r.usages as u (u.role + ':' + u.label + ':' + (u.role === 'hardcoded' ? u.path : ''))}
- {u.kind} + {#if u.kind === 'script'} + + {:else if u.kind === 'flow'} + + {:else} + + {/if} {u.label} + triggerDrawer?.closeDrawer()}> +
+

+ Triggers attached to the selected scripts and flows. Synced to the Hub as + disabled stubs. Recipients review and enable each one + manually after importing. External hooks (Slack/Discord webhooks, message-queue + subscriptions, etc.) must be re-registered against the importing instance. +

+ {#if relevantTriggers.length === 0} + No triggers reference the selected items. + {:else} + {#each triggersByKind as [kind, triggers] (kind)} +
+
+ + {TRIGGER_KIND_BADGE[kind]} + + + {triggers.length} trigger{triggers.length > 1 ? 's' : ''} + + {#if TRIGGER_KIND_NOTE[kind]} + + + + {#snippet text()} +
+ {TRIGGER_KIND_NOTE[kind]} +
+ {/snippet} + + + {/if} +
+
+ {#each triggers as t (t.path)} + {@const runnableSummary = runnableSummaryByPath.get( + `${t.is_flow ? 'flow' : 'script'}:${t.script_path}` + )} + {@const details = triggerDetails(t)} + {@const cfg = t.config as any} + {@const previewKey = t.kind === 'schedule' ? `${cfg.schedule}|${cfg.timezone}` : ''} + {@const preview = t.kind === 'schedule' ? schedulePreviews[previewKey] : undefined} +
+
+ {#if t.is_flow} + + {:else} + + {/if} + + {runnableSummary || t.script_path} + + + {t.is_flow ? 'flow' : 'script'} + +
+ {#if details.length > 0} +
+ {#each details as d (d.label)} +
{d.label}
+
{d.value}
+ {/each} + {#if t.kind === 'schedule'} +
Next runs
+
+ {#if preview && preview.length > 0} +
+ {#each preview as date (date)} + {dateFmt.format(new Date(date))} + {/each} +
+ {:else if preview && preview.length === 0} + No upcoming run + {:else} + Loading… + {/if} +
+ {/if} +
+ {/if} +
+ {/each} +
+
+ {/each} + {/if} +
+ + + bundleDrawer?.closeDrawer()}>
@@ -1558,6 +1985,24 @@ Name +
+ Project slug + + {effectiveSlug || sanitizeSlug(bundleName) || '—'} + + + {#if effectiveSlug} + Locked — items live under f/{effectiveSlug}/. + {:else if bundleName.trim() && !isValidSlug(sanitizeSlug(bundleName))} + + The name yields an invalid slug. Use at least 3 letters/digits. + + {:else} + Auto-generated from the name. Once project forked, items will live under + f/{sanitizeSlug(bundleName) || ''}/. + {/if} + +