feat: let agent workers authenticate with a workload identity token (#11858)

* feat: let agent workers authenticate with a workload identity token

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: throttle failed workload identity key fetches per issuer

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: keep stale workload identity keys through an issuer outage

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: refuse workload identity rules with an empty claim value

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: validate workload identity issuers in declarative sync and share the JWKS cache

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: say workload identity issuers are optional in agent worker settings

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* chore: update ee-repo-ref to 96e8e2aa15bdc5968a1dae10f4e81d47d7ea49e0

This commit updates the EE repository reference after PR #897 was merged in windmill-ee-private.

Previous ee-repo-ref: 0e98880edb8db269d53120ca606025c81f320f5c

New ee-repo-ref: 96e8e2aa15bdc5968a1dae10f4e81d47d7ea49e0

Automated by sync-ee-ref workflow.

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>

backend/Cargo.lock

@@ -15782,6 +15782,7 @@ dependencies = [
  "tracing",
  "uuid",
  "windmill-api-auth",
+ "windmill-audit",
  "windmill-common",
  "windmill-parser-py-imports",
  "windmill-queue",
@@ -16345,6 +16346,7 @@ dependencies = [
  "hex",
  "hkdf",
  "hmac 0.12.1",
+ "http 1.5.0",
  "hyper 1.12.0",
  "indexmap 2.14.2",
  "ipnet",

backend/ee-repo-ref.txt

@@ -1 +1 @@
-ca4a1959a0d2257d413a57349939053ac3473c6b
+96e8e2aa15bdc5968a1dae10f4e81d47d7ea49e0

backend/src/main.rs

@@ -686,6 +686,29 @@ async fn windmill_main() -> anyhow::Result<()> {
         _ => {}
     }
 
+    // Before `num_workers`: it reads `WORKER_GROUP`, which an agent authenticating with a
+    // workload identity token only learns from the server.
+    let agent_setup = if mode == Mode::Agent {
+        let agent_config = match AgentConfig::from_env() {
+            Ok(config) => config,
+            Err(e) => {
+                eprintln!("{e}");
+                std::process::exit(1);
+            }
+        };
+        let suffix = resolve_worker_suffix(&hostname, 1)?;
+        if let Err(e) = agent_config.resolve_workload_identity(&suffix).await {
+            eprintln!(
+                "Agent worker could not authenticate with its workload identity token, check \
+                 AGENT_IDENTITY_TOKEN_FILE and BASE_INTERNAL_URL (the reason is in the server log): {e:#}"
+            );
+            std::process::exit(1);
+        }
+        Some((agent_config, suffix))
+    } else {
+        None
+    };
+
     #[allow(unused_mut)]
     let mut num_workers = if mode == Mode::Server || mode == Mode::Indexer || mode == Mode::MCP {
         0
@@ -738,19 +761,11 @@ async fn windmill_main() -> anyhow::Result<()> {
         .and_then(|x| x.parse().ok())
         .unwrap_or(IpAddr::from(default_bind_addr));
 
-    let (conn, first_suffix, agent_config) = if mode == Mode::Agent {
-        let agent_config = match AgentConfig::from_env() {
-            Ok(config) => config,
-            Err(e) => {
-                tracing::error!("{e}");
-                std::process::exit(1);
-            }
-        };
+    let (conn, first_suffix, agent_config) = if let Some((agent_config, suffix)) = agent_setup {
         tracing::info!(
             "Creating http client for cluster using base internal url {}",
             agent_config.base_internal_url
         );
-        let suffix = resolve_worker_suffix(&hostname, 1)?;
         (
             Connection::Http(agent_config.build_http_client(&suffix)),
             Some(suffix),
@@ -996,7 +1011,7 @@ Windmill Community Edition {GIT_VERSION}
             tracing::error!("Failed to reload license key: {err:#}");
             if is_agent {
                 tracing::error!(
-                    "Agent worker cannot connect to server. Please check AGENT_TOKEN and BASE_INTERNAL_URL"
+                    "Agent worker cannot connect to server. Please check AGENT_TOKEN (or AGENT_IDENTITY_TOKEN_FILE) and BASE_INTERNAL_URL"
                 );
                 std::process::exit(1);
             }

backend/windmill-api-agent-workers/Cargo.toml

@@ -13,13 +13,14 @@ doctest = false
 [features]
 default = []
 enterprise = ["windmill-common/enterprise", "windmill-queue/enterprise"]
-private = ["windmill-common/private", "windmill-queue/private"]
+private = ["windmill-common/private", "windmill-queue/private", "windmill-audit/private"]
 python = ["dep:windmill-parser-py-imports"]
 benchmark = ["windmill-queue/benchmark"]
 
 [dependencies]
 windmill-api-auth.workspace = true
 windmill-common = { workspace = true, default-features = false }
+windmill-audit.workspace = true
 windmill-queue.workspace = true
 windmill-worker.workspace = true
 axum.workspace = true

backend/windmill-api-settings/src/lib.rs

@@ -1278,6 +1278,13 @@ async fn run_setting_pre_write_hook(
             v => windmill_common::global_settings::validate_accent_color(v)
                 .map_err(|e| error::Error::BadRequest(format!("{ACCENT_COLOR_SETTING}: {e}")))?,
         },
+        windmill_common::global_settings::AGENT_WORKLOAD_IDENTITY_ISSUERS_SETTING => {
+            if !value.is_null() {
+                let key = windmill_common::global_settings::AGENT_WORKLOAD_IDENTITY_ISSUERS_SETTING;
+                windmill_common::agent_workload_identity::validate_issuers_value(value)
+                    .map_err(|e| error::Error::BadRequest(format!("{key}: {e}")))?;
+            }
+        }
         _ => {}
     }
     Ok(())

backend/windmill-common/Cargo.toml

@@ -46,6 +46,7 @@ chrono.workspace = true
 chrono-tz.workspace = true
 hex.workspace = true
 async-trait.workspace = true
+http.workspace = true
 reqwest-middleware = { workspace = true }
 reqwest-retry = { workspace = true }
 rand.workspace = true

backend/windmill-common/src/agent_workers.rs

@@ -6,37 +6,83 @@ pub struct QueueInitJob {
 }
 
 use lazy_static::lazy_static;
-use std::time::Duration;
+use std::{path::PathBuf, time::Duration};
 
 use reqwest_middleware::ClientBuilder;
 use reqwest_retry::{policies::ExponentialBackoff, RetryTransientMiddleware};
 
 use crate::{jwt::decode_without_verify, utils::configure_client, worker::HttpClient};
 
-/// Configuration required for agent mode. Both fields are mandatory when running in agent mode.
+/// How an agent proves its identity to the server.
+#[derive(Clone)]
+pub enum AgentCredential {
+    /// `AGENT_TOKEN`: an agent token a superadmin minted.
+    Token(String),
+    /// `AGENT_IDENTITY_TOKEN_FILE`: a workload identity token the platform keeps rotating on
+    /// disk, re-read on every request.
+    IdentityTokenFile(PathBuf),
+}
+
+/// Configuration required for agent mode: a credential and the server's URL.
 #[derive(Clone)]
 pub struct AgentConfig {
-    pub agent_token: String,
+    pub credential: AgentCredential,
     pub base_internal_url: String,
 }
 
 impl AgentConfig {
     pub fn from_env() -> Result<Self, AgentConfigError> {
-        let agent_token =
-            std::env::var("AGENT_TOKEN").map_err(|_| AgentConfigError::MissingAgentToken)?;
+        let credential = match (
+            std::env::var("AGENT_TOKEN").ok(),
+            std::env::var(AGENT_IDENTITY_TOKEN_FILE_ENV).ok(),
+        ) {
+            (Some(_), Some(_)) => return Err(AgentConfigError::BothCredentials),
+            (Some(token), None) => AgentCredential::Token(token),
+            (None, Some(path)) => AgentCredential::IdentityTokenFile(PathBuf::from(path)),
+            (None, None) => return Err(AgentConfigError::MissingAgentToken),
+        };
         let base_internal_url = std::env::var("BASE_INTERNAL_URL")
             .map_err(|_| AgentConfigError::MissingBaseInternalUrl)?;
-        Ok(Self { agent_token, base_internal_url })
+        Ok(Self { credential, base_internal_url })
     }
 
     pub fn build_http_client(&self, worker_suffix: &str) -> HttpClient {
-        build_agent_http_client(worker_suffix, &self.agent_token, &self.base_internal_url)
+        match &self.credential {
+            AgentCredential::Token(token) => {
+                build_agent_http_client(worker_suffix, token, &self.base_internal_url)
+            }
+            AgentCredential::IdentityTokenFile(path) => {
+                build_agent_http_client_inner(&self.base_internal_url, None, |builder| {
+                    builder.with(IdentityTokenAuth {
+                        path: path.clone(),
+                        worker_suffix: worker_suffix.to_string(),
+                    })
+                })
+            }
+        }
+    }
+
+    /// With a workload identity token the worker group and tags are whatever rule the server
+    /// matched, so they are asked for before anything reads [`DECODED_AGENT_TOKEN`] (and
+    /// through it `WORKER_GROUP`). A no-op with a static agent token, which carries them.
+    pub async fn resolve_workload_identity(&self, worker_suffix: &str) -> anyhow::Result<()> {
+        if let AgentCredential::IdentityTokenFile(_) = self.credential {
+            let auth: AgentAuth = self
+                .build_http_client(worker_suffix)
+                .get("/api/agent_workers/whoami")
+                .await?;
+            let _ = AGENT_WORKLOAD_IDENTITY_AUTH.set(auth);
+        }
+        Ok(())
     }
 }
 
+pub const AGENT_IDENTITY_TOKEN_FILE_ENV: &str = "AGENT_IDENTITY_TOKEN_FILE";
+
 #[derive(Debug)]
 pub enum AgentConfigError {
     MissingAgentToken,
+    BothCredentials,
     MissingBaseInternalUrl,
 }
 
@@ -46,7 +92,13 @@ impl std::fmt::Display for AgentConfigError {
             AgentConfigError::MissingAgentToken => {
                 write!(
                     f,
-                    "AGENT_TOKEN environment variable is not set but required for agent mode"
+                    "AGENT_TOKEN or AGENT_IDENTITY_TOKEN_FILE environment variable is required for agent mode"
+                )
+            }
+            AgentConfigError::BothCredentials => {
+                write!(
+                    f,
+                    "AGENT_TOKEN and AGENT_IDENTITY_TOKEN_FILE are mutually exclusive, set only one"
                 )
             }
             AgentConfigError::MissingBaseInternalUrl => {
@@ -61,17 +113,56 @@ impl std::fmt::Display for AgentConfigError {
 
 impl std::error::Error for AgentConfigError {}
 
+/// Sets the bearer from the identity token file on each request, so a token the platform
+/// rotated on disk is picked up without a restart.
+struct IdentityTokenAuth {
+    path: PathBuf,
+    worker_suffix: String,
+}
+
+#[async_trait::async_trait]
+impl reqwest_middleware::Middleware for IdentityTokenAuth {
+    async fn handle(
+        &self,
+        mut req: reqwest::Request,
+        extensions: &mut http::Extensions,
+        next: reqwest_middleware::Next<'_>,
+    ) -> reqwest_middleware::Result<reqwest::Response> {
+        let token = tokio::fs::read_to_string(&self.path).await.map_err(|e| {
+            reqwest_middleware::Error::middleware(std::io::Error::new(
+                e.kind(),
+                format!(
+                    "reading {AGENT_IDENTITY_TOKEN_FILE_ENV} {}: {e}",
+                    self.path.display()
+                ),
+            ))
+        })?;
+        let header = format!(
+            "Bearer {AGENT_WORKLOAD_IDENTITY_PREFIX}{}{AGENT_WORKLOAD_IDENTITY_SEPARATOR}{}",
+            self.worker_suffix,
+            token.trim()
+        )
+        .parse()
+        .map_err(reqwest_middleware::Error::middleware)?;
+        req.headers_mut()
+            .insert(reqwest::header::AUTHORIZATION, header);
+        next.run(req, extensions).await
+    }
+}
+
 lazy_static! {
     pub static ref DECODED_AGENT_TOKEN: Option<AgentAuth> = {
         let agent_token = std::env::var("AGENT_TOKEN");
         if let Ok(token) = agent_token {
             decode_without_verify::<AgentAuth>(token.trim_start_matches(AGENT_JWT_PREFIX)).ok()
         } else {
-            None
+            AGENT_WORKLOAD_IDENTITY_AUTH.get().cloned()
         }
     };
 }
 
+static AGENT_WORKLOAD_IDENTITY_AUTH: std::sync::OnceLock<AgentAuth> = std::sync::OnceLock::new();
+
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct AgentAuth {
     pub worker_group: String,
@@ -81,11 +172,29 @@ pub struct AgentAuth {
 }
 
 pub const AGENT_JWT_PREFIX: &str = "jwt_agent_";
+/// `wid_agent_<worker suffix>:<workload identity JWT>`. A worker suffix may contain `_` (a
+/// `WORKER_SUFFIX` label) and a JWT may contain `_` (base64url) but neither contains `:`.
+pub const AGENT_WORKLOAD_IDENTITY_PREFIX: &str = "wid_agent_";
+pub const AGENT_WORKLOAD_IDENTITY_SEPARATOR: char = ':';
 
 pub fn build_agent_http_client(
     worker_suffix: &str,
     agent_token: &str,
     base_internal_url: &str,
+) -> HttpClient {
+    let token = format!(
+        "{}{}_{}",
+        AGENT_JWT_PREFIX,
+        worker_suffix,
+        agent_token.trim_start_matches(AGENT_JWT_PREFIX)
+    );
+    build_agent_http_client_inner(base_internal_url, Some(token), |builder| builder)
+}
+
+fn build_agent_http_client_inner(
+    base_internal_url: &str,
+    static_token: Option<String>,
+    with_auth: impl FnOnce(ClientBuilder) -> ClientBuilder,
 ) -> HttpClient {
     let client = ClientBuilder::new(
         configure_client(
@@ -101,16 +210,12 @@ pub fn build_agent_http_client(
                 "User-Agent",                          // Replace with your desired header name
                 "Windmill-Agent/1.0".parse().unwrap(), // Replace with your desired header value
             );
-            let token = format!(
-                "{}{}_{}",
-                AGENT_JWT_PREFIX,
-                worker_suffix,
-                agent_token.trim_start_matches(AGENT_JWT_PREFIX)
-            );
-            headers.insert(
-                "Authorization",
-                format!("Bearer {}", token).parse().unwrap(),
-            );
+            if let Some(token) = static_token {
+                headers.insert(
+                    "Authorization",
+                    format!("Bearer {}", token).parse().unwrap(),
+                );
+            }
             headers
         })
         .build()
@@ -118,10 +223,12 @@ pub fn build_agent_http_client(
     )
     .with(RetryTransientMiddleware::new_with_policy(
         ExponentialBackoff::builder().build_with_max_retries(5),
-    ))
-    .build();
+    ));
 
-    HttpClient { client, base_internal_url: base_internal_url.to_string() }
+    HttpClient {
+        client: with_auth(client).build(),
+        base_internal_url: base_internal_url.to_string(),
+    }
 }
 
 #[derive(Deserialize, Serialize)]

backend/windmill-common/src/agent_workload_identity.rs

@@ -0,0 +1,475 @@
+//! Agent worker authentication with a workload identity token: a JWT the platform the agent
+//! runs on issues to it (a Kubernetes projected service account token, a SPIFFE JWT-SVID, ...),
+//! verified against an issuer the superadmin trusts in `agent_workload_identity_issuers`.
+//! The worker group and tags come from the first rule whose claims the token matches, never
+//! from the token's own say, so the tag checks the static agent token path applies hold as is.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use jsonwebtoken::{jwk::Jwk, Algorithm, DecodingKey, Validation};
+use serde_json::Value;
+
+use crate::error::Result;
+use crate::guest_jwt::{cached_jwks_keys, jwk_algorithms, parse_jwks_keys, JwksSource};
+use crate::instance_config::{AgentWorkloadIdentityIssuer, AgentWorkloadIdentityRule};
+
+/// A projected service account token is about 1 KB; refused past this before any work.
+const MAX_TOKEN_LEN: usize = 16 * 1024;
+
+/// Why a token was refused. `issuer` is set once the token named a trusted one. The reason
+/// goes to the server log and the audit log, never to the client.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Refusal {
+    pub issuer: Option<String>,
+    /// The token's `sub`, set only once its signature verified, so a caller cannot vary it.
+    pub verified_subject: Option<String>,
+    pub reason: String,
+}
+
+impl Refusal {
+    fn new(issuer: Option<&str>, reason: impl Into<String>) -> Self {
+        Self { issuer: issuer.map(str::to_string), verified_subject: None, reason: reason.into() }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WorkloadIdentity {
+    pub issuer: String,
+    pub subject: Option<String>,
+    pub worker_group: String,
+    pub tags: Vec<String>,
+    pub exp: u64,
+}
+
+/// The setting's raw value, as both the settings API and the declarative sync receive it.
+pub fn validate_issuers_value(value: &Value) -> std::result::Result<(), String> {
+    let issuers: Vec<AgentWorkloadIdentityIssuer> =
+        serde_json::from_value(value.clone()).map_err(|e| e.to_string())?;
+    validate_issuers(&issuers)
+}
+
+/// Rejects what would make a rule set ambiguous or a rule match more than intended. Runtime
+/// matching skips a rule without claims too, so a setting planted around this check cannot
+/// admit every token of an issuer.
+pub fn validate_issuers(
+    issuers: &[AgentWorkloadIdentityIssuer],
+) -> std::result::Result<(), String> {
+    let mut seen = std::collections::HashSet::new();
+    for (i, cfg) in issuers.iter().enumerate() {
+        let at = format!("issuer #{} ({})", i + 1, cfg.issuer);
+        if cfg.issuer.trim().is_empty() {
+            return Err(format!("issuer #{}: issuer is required", i + 1));
+        }
+        if !seen.insert(cfg.issuer.as_str()) {
+            return Err(format!("{at}: listed more than once"));
+        }
+        if cfg.audience.trim().is_empty() {
+            return Err(format!("{at}: audience is required"));
+        }
+        match (&cfg.jwks, &cfg.jwks_uri) {
+            (Some(jwks), _) if !jwks.trim().is_empty() => {
+                parse_jwks_keys(jwks.as_bytes()).map_err(|e| format!("{at}: inline JWKS: {e}"))?;
+            }
+            (_, Some(uri)) if !uri.trim().is_empty() => {
+                if !(uri.starts_with("https://") || uri.starts_with("http://")) {
+                    return Err(format!("{at}: jwks_uri must be an http(s) URL"));
+                }
+            }
+            _ => {
+                if !(cfg.issuer.starts_with("https://") || cfg.issuer.starts_with("http://")) {
+                    return Err(format!(
+                        "{at}: without jwks_uri or an inline JWKS, the issuer must be an http(s) URL \
+                         serving OIDC discovery"
+                    ));
+                }
+            }
+        }
+        if cfg.rules.is_empty() {
+            return Err(format!("{at}: at least one rule is required"));
+        }
+        for (j, rule) in cfg.rules.iter().enumerate() {
+            let at = format!("{at}, rule #{}", j + 1);
+            if rule.claims.is_empty()
+                || rule
+                    .claims
+                    .iter()
+                    .any(|(k, v)| k.trim().is_empty() || v.trim().is_empty())
+            {
+                return Err(format!(
+                    "{at}: at least one claim to match is required, each with a path and a value"
+                ));
+            }
+            if rule.worker_group.trim().is_empty()
+                || rule.worker_group.contains(char::is_whitespace)
+            {
+                return Err(format!("{at}: worker group is required and has no spaces"));
+            }
+            if rule.tags.is_empty() || rule.tags.iter().any(|t| t.trim().is_empty()) {
+                return Err(format!("{at}: at least one non-empty tag is required"));
+            }
+        }
+    }
+    Ok(())
+}
+
+/// The value at a dot separated claim path. At each level the longest key that prefixes the
+/// remaining path is tried first, so `kubernetes.io.namespace` reaches `namespace` under the
+/// `kubernetes.io` key.
+fn claim_at<'a>(value: &'a Value, path: &str) -> Option<&'a Value> {
+    let obj = value.as_object()?;
+    if let Some(v) = obj.get(path) {
+        return Some(v);
+    }
+    path.match_indices('.')
+        .rev()
+        .find_map(|(i, _)| claim_at(obj.get(&path[..i])?, &path[i + 1..]))
+}
+
+fn claim_matches(claims: &Value, path: &str, expected: &str) -> bool {
+    match claim_at(claims, path) {
+        Some(Value::String(s)) => s == expected,
+        Some(v @ (Value::Number(_) | Value::Bool(_))) => v.to_string() == expected,
+        _ => false,
+    }
+}
+
+pub fn matching_rule<'a>(
+    rules: &'a [AgentWorkloadIdentityRule],
+    claims: &Value,
+) -> Option<&'a AgentWorkloadIdentityRule> {
+    rules.iter().find(|rule| {
+        !rule.claims.is_empty()
+            && rule
+                .claims
+                .iter()
+                .all(|(path, expected)| claim_matches(claims, path, expected))
+    })
+}
+
+/// The header's `alg`, read raw: `jsonwebtoken` cannot parse `none`, and a refusal should say
+/// which algorithm was offered rather than that the header is malformed.
+fn header_alg(token: &str) -> std::result::Result<Algorithm, String> {
+    use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
+    let header = token.split('.').next().unwrap_or_default();
+    let header: Value = URL_SAFE_NO_PAD
+        .decode(header.trim_end_matches('='))
+        .ok()
+        .and_then(|b| serde_json::from_slice(&b).ok())
+        .ok_or("malformed JWT header")?;
+    let alg = header.get("alg").and_then(Value::as_str).unwrap_or("");
+    match alg.parse::<Algorithm>() {
+        Ok(
+            a @ (Algorithm::RS256
+            | Algorithm::RS384
+            | Algorithm::RS512
+            | Algorithm::PS256
+            | Algorithm::PS384
+            | Algorithm::PS512
+            | Algorithm::ES256
+            | Algorithm::ES384),
+        ) => Ok(a),
+        _ => Err(format!(
+            "alg {alg:?} is not accepted, only RS*, PS* and ES*"
+        )),
+    }
+}
+
+/// Verify `token` against the trusted `issuers` and map it to a worker group and tags.
+pub async fn verify(
+    issuers: &[AgentWorkloadIdentityIssuer],
+    token: &str,
+) -> std::result::Result<WorkloadIdentity, Refusal> {
+    if token.len() > MAX_TOKEN_LEN {
+        return Err(Refusal::new(None, "token is too long"));
+    }
+    let alg = header_alg(token).map_err(|r| Refusal::new(None, r))?;
+    let kid = jsonwebtoken::decode_header(token)
+        .map_err(|e| Refusal::new(None, format!("malformed JWT header: {e}")))?
+        .kid;
+    let unverified: Value = crate::jwt::decode_without_verify(token)
+        .map_err(|e| Refusal::new(None, format!("malformed JWT payload: {e}")))?;
+    let iss = unverified.get("iss").and_then(Value::as_str).unwrap_or("");
+    let cfg = issuers
+        .iter()
+        .find(|c| c.issuer == iss)
+        .ok_or_else(|| Refusal::new(None, format!("issuer {iss:?} is not trusted")))?;
+    let refuse = |reason: String| Refusal::new(Some(&cfg.issuer), reason);
+
+    let keys = issuer_keys(cfg)
+        .await
+        .map_err(|e| refuse(format!("signing keys unavailable: {e}")))?;
+    let candidates: Vec<&Jwk> = match &kid {
+        Some(kid) => keys.get(kid).into_iter().collect(),
+        None => keys.values().collect(),
+    };
+    let candidates: Vec<&Jwk> = candidates
+        .into_iter()
+        .filter(|jwk| jwk_algorithms(jwk).is_some_and(|algs| algs.contains(&alg)))
+        .collect();
+    if candidates.is_empty() {
+        return Err(refuse(match &kid {
+            Some(kid) => format!("no {alg:?} signing key with kid {kid:?} in the issuer's JWKS"),
+            None => format!("no {alg:?} signing key in the issuer's JWKS"),
+        }));
+    }
+
+    let mut validation = Validation::new(alg);
+    validation.set_audience(&[&cfg.audience]);
+    validation.set_issuer(&[&cfg.issuer]);
+    validation.set_required_spec_claims(&["exp", "iss", "aud"]);
+    validation.validate_nbf = true;
+    let mut last_err = None;
+    let mut claims = None;
+    for jwk in candidates {
+        let key = match DecodingKey::from_jwk(jwk) {
+            Ok(key) => key,
+            Err(e) => {
+                last_err = Some(e.to_string());
+                continue;
+            }
+        };
+        match jsonwebtoken::decode::<Value>(token, &key, &validation) {
+            Ok(data) => {
+                claims = Some(data.claims);
+                break;
+            }
+            Err(e) => last_err = Some(describe_jwt_error(&e, &cfg.audience)),
+        }
+    }
+    let claims = claims.ok_or_else(|| refuse(last_err.unwrap_or_default()))?;
+
+    let subject = claims
+        .get("sub")
+        .and_then(Value::as_str)
+        .map(str::to_string);
+    let rule = matching_rule(&cfg.rules, &claims).ok_or_else(|| Refusal {
+        verified_subject: subject.clone(),
+        ..refuse(format!(
+            "no rule matches the token (sub {})",
+            subject.as_deref().unwrap_or("<none>")
+        ))
+    })?;
+    Ok(WorkloadIdentity {
+        issuer: cfg.issuer.clone(),
+        subject,
+        worker_group: rule.worker_group.clone(),
+        tags: rule.tags.clone(),
+        exp: claims.get("exp").and_then(Value::as_u64).unwrap_or(0),
+    })
+}
+
+fn describe_jwt_error(e: &jsonwebtoken::errors::Error, audience: &str) -> String {
+    use jsonwebtoken::errors::ErrorKind;
+    match e.kind() {
+        ErrorKind::InvalidSignature => "signature does not verify".to_string(),
+        ErrorKind::ExpiredSignature => "token is expired".to_string(),
+        ErrorKind::ImmatureSignature => "token is not valid yet (nbf)".to_string(),
+        ErrorKind::InvalidAudience => format!("aud does not include {audience:?}"),
+        ErrorKind::InvalidIssuer => "iss does not match".to_string(),
+        ErrorKind::MissingRequiredClaim(c) => format!("required claim {c} is missing"),
+        _ => format!("token refused: {e}"),
+    }
+}
+
+async fn issuer_keys(cfg: &AgentWorkloadIdentityIssuer) -> Result<Arc<HashMap<String, Jwk>>> {
+    if let Some(jwks) = cfg.jwks.as_deref().filter(|j| !j.trim().is_empty()) {
+        return Ok(Arc::new(parse_jwks_keys(jwks.as_bytes())?));
+    }
+    cached_jwks_keys(&JwksSource::TrustedIssuer {
+        issuer: cfg.issuer.clone(),
+        jwks_uri: cfg.jwks_uri.clone(),
+    })
+    .await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use jsonwebtoken::{EncodingKey, Header};
+    use serde_json::json;
+    use std::collections::BTreeMap;
+
+    // Throwaway P-256 keys, used nowhere else. `JWKS` publishes the first one.
+    const PRIV: &str = "-----BEGIN PRIVATE KEY-----
+MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgu7FqG4gEcV8logVI
+B6/Lc5aFA8jxKmix7Jdcr/acid2hRANCAATV+8XhZlMNXkRAy7FZ4nbQkBOechNv
+a2U/COEBOOauFRMd0NB5bi2DAqO1r/h7eN7fLMGYLXVPlSvNdJuYL4F8
+-----END PRIVATE KEY-----";
+    const OTHER_PRIV: &str = "-----BEGIN PRIVATE KEY-----
+MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgcU92tN6g4nOPoxQm
+6KLBG0mtRL4gZ6/BNcrNPXiRammhRANCAASu64yo/1u80XgfncT4r9QMORMHZjDM
+Sa6vBkm9YSs4MMV6K5vCIyow+i712HgUL/FNZ55GJVwXF77JffommMAk
+-----END PRIVATE KEY-----";
+    const JWKS: &str = r#"{"keys":[{"kty":"EC","crv":"P-256","alg":"ES256","use":"sig","kid":"k1",
+        "x":"1fvF4WZTDV5EQMuxWeJ20JATnnITb2tlPwjhATjmrhU",
+        "y":"Ex3Q0HluLYMCo7Wv-Ht43t8swZgtdU-VK810m5gvgXw"}]}"#;
+    const ISS: &str = "https://kubernetes.default.svc.cluster.local";
+
+    fn issuer() -> AgentWorkloadIdentityIssuer {
+        AgentWorkloadIdentityIssuer {
+            issuer: ISS.to_string(),
+            jwks_uri: None,
+            jwks: Some(JWKS.to_string()),
+            audience: "windmill".to_string(),
+            rules: vec![
+                AgentWorkloadIdentityRule {
+                    claims: BTreeMap::from([(
+                        "sub".to_string(),
+                        "system:serviceaccount:agents:gpu".to_string(),
+                    )]),
+                    worker_group: "gpu".to_string(),
+                    tags: vec!["gpu".to_string()],
+                },
+                AgentWorkloadIdentityRule {
+                    claims: BTreeMap::from([(
+                        "kubernetes.io.namespace".to_string(),
+                        "agents".to_string(),
+                    )]),
+                    worker_group: "agents".to_string(),
+                    tags: vec!["dbt".to_string(), "etl".to_string()],
+                },
+            ],
+        }
+    }
+
+    fn claims(sub: &str, ns: &str) -> Value {
+        let now = jsonwebtoken::get_current_timestamp();
+        json!({
+            "iss": ISS, "aud": ["windmill"], "sub": sub, "exp": now + 600, "iat": now, "nbf": now,
+            "kubernetes.io": {"namespace": ns, "serviceaccount": {"name": "x"}},
+        })
+    }
+
+    fn sign(claims: &Value, pem: &str, alg: Algorithm) -> String {
+        let mut header = Header::new(alg);
+        header.kid = Some("k1".to_string());
+        jsonwebtoken::encode(
+            &header,
+            claims,
+            &EncodingKey::from_ec_pem(pem.as_bytes()).unwrap(),
+        )
+        .unwrap()
+    }
+
+    async fn reason(token: &str) -> String {
+        verify(&[issuer()], token).await.unwrap_err().reason
+    }
+
+    #[tokio::test]
+    async fn first_matching_rule_decides_group_and_tags() {
+        let id = verify(
+            &[issuer()],
+            &sign(
+                &claims("system:serviceaccount:agents:gpu", "agents"),
+                PRIV,
+                Algorithm::ES256,
+            ),
+        )
+        .await
+        .unwrap();
+        assert_eq!(
+            (id.worker_group.as_str(), id.tags),
+            ("gpu", vec!["gpu".to_string()])
+        );
+        // A dotted key (`kubernetes.io`) is reached through a nested path.
+        let id = verify(
+            &[issuer()],
+            &sign(
+                &claims("system:serviceaccount:agents:etl", "agents"),
+                PRIV,
+                Algorithm::ES256,
+            ),
+        )
+        .await
+        .unwrap();
+        assert_eq!(id.worker_group, "agents");
+        assert!(reason(&sign(
+            &claims("system:serviceaccount:other:etl", "other"),
+            PRIV,
+            Algorithm::ES256
+        ))
+        .await
+        .starts_with("no rule matches"));
+    }
+
+    #[tokio::test]
+    async fn refuses_wrong_iss_aud_exp_key_and_alg() {
+        let ok = claims("system:serviceaccount:agents:gpu", "agents");
+        let with = |k: &str, v: Value| {
+            let mut c = ok.clone();
+            c[k] = v;
+            sign(&c, PRIV, Algorithm::ES256)
+        };
+        assert!(reason(&with("iss", json!("https://evil")))
+            .await
+            .contains("not trusted"));
+        assert!(reason(&with("aud", json!("other"))).await.contains("aud"));
+        assert!(reason(&with("exp", json!(1_000_000)))
+            .await
+            .contains("expired"));
+        assert!(reason(&sign(&ok, OTHER_PRIV, Algorithm::ES256))
+            .await
+            .contains("signature"));
+
+        let hs = jsonwebtoken::encode(
+            &Header::new(Algorithm::HS256),
+            &ok,
+            &EncodingKey::from_secret(b"s"),
+        )
+        .unwrap();
+        assert!(reason(&hs).await.contains("not accepted"));
+        use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
+        let none = format!(
+            "{}.{}.",
+            URL_SAFE_NO_PAD.encode(br#"{"alg":"none","kid":"k1"}"#),
+            URL_SAFE_NO_PAD.encode(ok.to_string())
+        );
+        assert!(reason(&none).await.contains("not accepted"));
+    }
+
+    #[tokio::test]
+    async fn failed_key_fetches_are_throttled() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let hits = Arc::new(AtomicUsize::new(0));
+        let counted = hits.clone();
+        tokio::spawn(async move {
+            while let Ok((conn, _)) = listener.accept().await {
+                counted.fetch_add(1, Ordering::SeqCst);
+                drop(conn);
+            }
+        });
+        let mut cfg = issuer();
+        cfg.issuer = format!("https://throttle-test-{}", addr.port());
+        cfg.jwks = None;
+        cfg.jwks_uri = Some(format!("http://{addr}/jwks"));
+        let mut c = claims("system:serviceaccount:agents:gpu", "agents");
+        c["iss"] = json!(cfg.issuer);
+        for _ in 0..3 {
+            let refusal = verify(&[cfg.clone()], &sign(&c, PRIV, Algorithm::ES256))
+                .await
+                .unwrap_err();
+            assert!(
+                refusal.reason.contains("signing keys unavailable"),
+                "{}",
+                refusal.reason
+            );
+        }
+        assert_eq!(hits.load(Ordering::SeqCst), 1);
+    }
+
+    #[test]
+    fn rule_without_claims_never_matches() {
+        let rules = vec![AgentWorkloadIdentityRule {
+            claims: BTreeMap::new(),
+            worker_group: "g".to_string(),
+            tags: vec!["t".to_string()],
+        }];
+        assert!(matching_rule(&rules, &json!({"sub": "x"})).is_none());
+        let mut cfg = issuer();
+        cfg.rules = rules;
+        assert!(validate_issuers(&[cfg]).is_err());
+    }
+}

backend/windmill-common/src/global_settings.rs

@@ -68,6 +68,7 @@ pub const EXTERNAL_INSTANCE_PG_SETTING: &str = "external_instance_pg";
 pub const INSTANCE_PG_DISABLED_SETTING: &str = "instance_pg_disabled";
 pub const EXTERNAL_INSTANCE_PG_STATE_SETTING: &str = "external_instance_pg_state";
 pub const TIMEOUT_WAIT_RESULT_SETTING: &str = "timeout_wait_result";
+pub const AGENT_WORKLOAD_IDENTITY_ISSUERS_SETTING: &str = "agent_workload_identity_issuers";
 
 pub const UNIQUE_ID_SETTING: &str = "uid";
 pub const DISABLE_STATS_SETTING: &str = "disable_stats";

backend/windmill-common/src/guest_jwt.rs

@@ -328,7 +328,6 @@ pub const MAX_GUEST_JWT_LEN: usize = 8 * 1024;
 /// self-hosted internal (http/private) issuer that works for `jwt_ext_` works for guests too.
 /// The body is read with a cap so a hostile endpoint cannot exhaust memory.
 pub async fn fetch_jwks(url: &str) -> Result<HashMap<String, Jwk>> {
-    use futures::StreamExt;
     if url.len() > MAX_JWKS_URL_LEN {
         return Err(Error::BadRequest(format!(
             "JWKS URL is longer than {MAX_JWKS_URL_LEN} bytes"
@@ -356,6 +355,13 @@ pub async fn fetch_jwks(url: &str) -> Result<HashMap<String, Jwk>> {
     }
     .and_then(|r| r.error_for_status())
     .map_err(|e| Error::BadRequest(format!("could not fetch JWKS: {e}")))?;
+    parse_jwks_keys(&read_capped_body(resp).await?)
+}
+
+/// A JWKS (or OIDC discovery) response body, refused past `JWKS_MAX_BYTES` rather than
+/// buffered: the server behind a configured URL may be hostile.
+pub(crate) async fn read_capped_body(resp: reqwest::Response) -> Result<Vec<u8>> {
+    use futures::StreamExt;
     let mut stream = resp.bytes_stream();
     let mut body: Vec<u8> = Vec::new();
     while let Some(chunk) = stream.next().await {
@@ -367,7 +373,7 @@ pub async fn fetch_jwks(url: &str) -> Result<HashMap<String, Jwk>> {
         }
         body.extend_from_slice(&chunk);
     }
-    parse_jwks_keys(&body)
+    Ok(body)
 }
 
 /// The usable signing keys in a JWKS body, by `kid`. Each key is parsed on its own and
@@ -378,7 +384,7 @@ pub async fn fetch_jwks(url: &str) -> Result<HashMap<String, Jwk>> {
 /// A key is kept only if its material actually decodes (`DecodingKey::from_jwk`): jsonwebtoken
 /// carries `n`/`e`/`x`/`y` as strings and defers decoding to auth time, so without this a JWKS
 /// whose only key is malformed would be accepted at save time and fail every token later.
-fn parse_jwks_keys(body: &[u8]) -> Result<HashMap<String, Jwk>> {
+pub(crate) fn parse_jwks_keys(body: &[u8]) -> Result<HashMap<String, Jwk>> {
     let set: serde_json::Value = serde_json::from_slice(body)
         .map_err(|e| Error::BadRequest(format!("JWKS is not JSON: {e}")))?;
     let entries = set
@@ -422,13 +428,75 @@ fn parse_jwks_keys(body: &[u8]) -> Result<HashMap<String, Jwk>> {
 fn servable(entry: Arc<JwksEntry>) -> Result<Arc<JwksEntry>> {
     if entry.keys.is_empty() {
         Err(Error::NotAuthorized(
-            "guest JWT refused: the JWKS issuer is unreachable".to_string(),
+            "the JWKS issuer is unreachable".to_string(),
         ))
     } else {
         Ok(entry)
     }
 }
 
+/// Where a cached JWKS comes from. Both kinds share the cache, its single-flight refresh and
+/// its staleness bounds; they differ only in how a fetch reaches the issuer.
+#[derive(Clone)]
+pub enum JwksSource {
+    /// A workspace guest key URL: SSRF-restricted unless it is the instance issuer
+    /// (`fetch_jwks`).
+    GuestUrl(String),
+    /// An issuer a superadmin trusts for agent workload identity. Trusted like
+    /// `JWT_EXT_JWKS_URL`, so fetched with the permissive client: an in-cluster or
+    /// self-signed issuer has to work. Without `jwks_uri`, OIDC discovery finds it.
+    TrustedIssuer { issuer: String, jwks_uri: Option<String> },
+}
+
+impl JwksSource {
+    fn cache_key(&self) -> String {
+        match self {
+            JwksSource::GuestUrl(url) => url.clone(),
+            JwksSource::TrustedIssuer { issuer, jwks_uri } => {
+                format!("issuer:{issuer}\n{}", jwks_uri.as_deref().unwrap_or(""))
+            }
+        }
+    }
+
+    async fn fetch(&self) -> Result<HashMap<String, Jwk>> {
+        match self {
+            JwksSource::GuestUrl(url) => fetch_jwks(url).await,
+            JwksSource::TrustedIssuer { issuer, jwks_uri } => {
+                let jwks_uri = match jwks_uri.as_deref().filter(|u| !u.trim().is_empty()) {
+                    Some(uri) => uri.to_string(),
+                    None => {
+                        let discovery = format!(
+                            "{}/.well-known/openid-configuration",
+                            issuer.trim_end_matches('/')
+                        );
+                        let doc: serde_json::Value =
+                            serde_json::from_slice(&get_trusted(&discovery).await?).map_err(
+                                |e| Error::BadRequest(format!("OIDC discovery is not JSON: {e}")),
+                            )?;
+                        doc.get("jwks_uri")
+                            .and_then(|u| u.as_str())
+                            .ok_or_else(|| {
+                                Error::BadRequest("OIDC discovery has no jwks_uri".to_string())
+                            })?
+                            .to_string()
+                    }
+                };
+                parse_jwks_keys(&get_trusted(&jwks_uri).await?)
+            }
+        }
+    }
+}
+
+async fn get_trusted(url: &str) -> Result<Vec<u8>> {
+    let resp = crate::utils::HTTP_CLIENT_PERMISSIVE
+        .get(url)
+        .send()
+        .await
+        .and_then(|r| r.error_for_status())
+        .map_err(|e| Error::BadRequest(format!("fetching {url}: {e}")))?;
+    read_capped_body(resp).await
+}
+
 /// A held single-flight lock for one JWKS URL. Dropping it removes the URL from
 /// `JWKS_FETCH_LOCKS` once no other holder remains, so the registry never keeps a lock past
 /// its fetch and stays bounded by the number of fetches in flight, not by URLs ever seen.
@@ -468,13 +536,14 @@ impl Drop for JwksFetchLock {
 /// Refresh a URL's JWKS off the request path, under the single-flight lock. A held
 /// lock means a refresh is already running, so this is a no-op. A failed refresh
 /// leaves the served stale keys in place rather than dropping them.
-fn spawn_jwks_refresh(url: String) {
+fn spawn_jwks_refresh(source: JwksSource) {
     tokio::spawn(async move {
+        let url = source.cache_key();
         let fetch_lock = JwksFetchLock::acquire(&url);
         let Ok(_guard) = fetch_lock.lock.try_lock() else {
             return;
         };
-        match fetch_jwks(&url).await {
+        match source.fetch().await {
             Ok(keys) => {
                 let now = Instant::now();
                 JWKS_CACHE.insert(
@@ -486,7 +555,7 @@ fn spawn_jwks_refresh(url: String) {
                     }),
                 );
             }
-            Err(e) => tracing::warn!("guest JWKS background refresh failed for {url}: {e:#}"),
+            Err(e) => tracing::warn!("JWKS background refresh failed for {url}: {e:#}"),
         }
     });
 }
@@ -498,7 +567,9 @@ fn spawn_jwks_refresh(url: String) {
 /// entry that reads as "issuer unreachable", so an unreachable issuer is hit at most
 /// once per `JWKS_NEGATIVE_TTL`. Fetches follow a schedule, never a per-request,
 /// attacker-chosen `kid`.
-async fn cached_jwks(url: &str) -> Result<Arc<JwksEntry>> {
+async fn cached_jwks(source: &JwksSource) -> Result<Arc<JwksEntry>> {
+    let key = source.cache_key();
+    let url = key.as_str();
     if let Some(entry) = JWKS_CACHE.get(url) {
         // Keys past JWKS_MAX_STALE are never served, even while revalidating: a stale re-serve
         // bumps expires_at but keeps fetched_at, so a persistently failing refresh would
@@ -519,7 +590,7 @@ async fn cached_jwks(url: &str) -> Result<Arc<JwksEntry>> {
                     fetched_at: entry.fetched_at,
                 });
                 JWKS_CACHE.insert(url.to_string(), served.clone());
-                spawn_jwks_refresh(url.to_string());
+                spawn_jwks_refresh(source.clone());
                 return Ok(served);
             }
         }
@@ -535,7 +606,7 @@ async fn cached_jwks(url: &str) -> Result<Arc<JwksEntry>> {
             return servable(entry);
         }
     }
-    match fetch_jwks(url).await {
+    match source.fetch().await {
         Ok(keys) => {
             let now = Instant::now();
             let entry = Arc::new(JwksEntry {
@@ -573,7 +644,12 @@ pub async fn jwks_key_for(url: &str, token: &str) -> Result<(DecodingKey, Vec<Al
     let kid = header.kid.ok_or_else(|| {
         Error::NotAuthorized("guest JWT refused: no kid in the header".to_string())
     })?;
-    let entry = cached_jwks(url).await?;
+    let entry = cached_jwks(&JwksSource::GuestUrl(url.to_string()))
+        .await
+        .map_err(|e| match e {
+            Error::NotAuthorized(m) => Error::NotAuthorized(format!("guest JWT refused: {m}")),
+            e => e,
+        })?;
     let jwk = entry.keys.get(&kid).ok_or_else(|| {
         Error::NotAuthorized(format!("guest JWT refused: kid {kid} is not in the JWKS"))
     })?;
@@ -585,6 +661,12 @@ pub async fn jwks_key_for(url: &str, token: &str) -> Result<(DecodingKey, Vec<Al
     Ok((key, algs))
 }
 
+/// The keys `source` currently serves, through the shared cache. An unknown `kid` is refused
+/// against them by the caller, never refetched, for the reason `jwks_key_for` gives.
+pub async fn cached_jwks_keys(source: &JwksSource) -> Result<Arc<HashMap<String, Jwk>>> {
+    Ok(cached_jwks(source).await?.keys.clone())
+}
+
 /// Verify `token` for `w_id` against whatever key the workspace configured. A PEM key
 /// ignores `kid`; a JWKS selects by it.
 pub async fn verify_for_workspace(db: &DB, w_id: &str, token: &str) -> Result<GuestJwtClaims> {
@@ -950,7 +1032,9 @@ y9rTR828ADcaZ63Ej1oL4GcqmGhODxCLy1YKKcy0FHzChqPMV6g=\n\
         for _ in 0..10 {
             let u = url.clone();
             handles.push(tokio::spawn(async move {
-                cached_jwks(&u).await.map(|e| e.keys.len())
+                cached_jwks(&JwksSource::GuestUrl(u))
+                    .await
+                    .map(|e| e.keys.len())
             }));
         }
         for h in handles {
@@ -1013,7 +1097,7 @@ y9rTR828ADcaZ63Ej1oL4GcqmGhODxCLy1YKKcy0FHzChqPMV6g=\n\
             }),
         );
         assert!(
-            cached_jwks(&within).await.is_ok(),
+            cached_jwks(&JwksSource::GuestUrl(within)).await.is_ok(),
             "stale keys within the grace window are still served"
         );
 
@@ -1030,7 +1114,7 @@ y9rTR828ADcaZ63Ej1oL4GcqmGhODxCLy1YKKcy0FHzChqPMV6g=\n\
             }),
         );
         assert!(
-            cached_jwks(&beyond).await.is_err(),
+            cached_jwks(&JwksSource::GuestUrl(beyond)).await.is_err(),
             "keys past the grace window fail closed once the refresh fails"
         );
         unsafe { std::env::remove_var("ALLOW_PRIVATE_GUEST_JWKS_URLS") };

backend/windmill-common/src/instance_config.rs

@@ -368,6 +368,8 @@ pub struct GlobalSettings {
     pub custom_instance_pg_databases: Option<CustomInstancePgDatabases>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub external_instance_pg: Option<ExternalInstancePg>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub agent_workload_identity_issuers: Option<Vec<AgentWorkloadIdentityIssuer>>,
 
     // Opaque settings (EE-private structs or no clear schema)
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -738,6 +740,42 @@ pub struct TeamsChannel {
     pub channel_name: String,
 }
 
+// ---------------------------------------------------------------------------
+// Agent worker workload identity
+// ---------------------------------------------------------------------------
+
+/// An issuer whose workload identity tokens (Kubernetes projected service account tokens,
+/// SPIFFE JWT-SVIDs, ...) an agent worker may authenticate with instead of an agent token.
+#[derive(Deserialize, Serialize, Clone, Debug, Default, PartialEq)]
+#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
+pub struct AgentWorkloadIdentityIssuer {
+    /// Must equal the token's `iss` exactly.
+    pub issuer: String,
+    /// Where the signing keys are published. Defaults to the `jwks_uri` of the issuer's OIDC
+    /// discovery document.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub jwks_uri: Option<String>,
+    /// The JWKS document itself, as JSON text, for an issuer this instance cannot reach.
+    /// Takes precedence over `jwks_uri`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub jwks: Option<String>,
+    /// Must be one of the token's `aud`.
+    pub audience: String,
+    /// Tried in order; the first whose claims all match decides the worker group and tags.
+    #[serde(default)]
+    pub rules: Vec<AgentWorkloadIdentityRule>,
+}
+
+#[derive(Deserialize, Serialize, Clone, Debug, Default, PartialEq)]
+#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
+pub struct AgentWorkloadIdentityRule {
+    /// Claim path (dot separated, a key that itself contains dots such as `kubernetes.io`
+    /// is matched whole) to the exact value it must hold.
+    pub claims: BTreeMap<String, String>,
+    pub worker_group: String,
+    pub tags: Vec<String>,
+}
+
 // ---------------------------------------------------------------------------
 // DB oversize alert
 // ---------------------------------------------------------------------------
@@ -1508,6 +1546,13 @@ pub async fn sync_global_settings_declarative(
             .map_err(|e| anyhow::anyhow!("{accent_key}: {e}"))?,
     }
 
+    let issuers_key = crate::global_settings::AGENT_WORKLOAD_IDENTITY_ISSUERS_SETTING;
+    match desired.get(issuers_key) {
+        None | Some(serde_json::Value::Null) => {}
+        Some(issuers) => crate::agent_workload_identity::validate_issuers_value(issuers)
+            .map_err(|e| anyhow::anyhow!("{issuers_key}: {e}"))?,
+    }
+
     let suffix_key = crate::global_settings::DEV_WORKSPACE_TAG_SUFFIX_SETTING;
     match desired.get(suffix_key) {
         None | Some(serde_json::Value::Null) => {}

backend/windmill-common/src/lib.rs

@@ -27,6 +27,7 @@ use scripts::ScriptLang;
 use sqlx::{Acquire, Postgres};
 
 pub mod agent_workers;
+pub mod agent_workload_identity;
 pub mod apps;
 pub mod assets;
 pub mod audit;

backend/windmill-common/src/utils.rs

@@ -124,8 +124,10 @@ lazy_static::lazy_static! {
                 if std::env::var("BASE_INTERNAL_URL").is_err() {
                     panic!("BASE_INTERNAL_URL is required in agent mode")
                 }
-                if std::env::var("AGENT_TOKEN").is_err() {
-                    println!("AGENT_TOKEN is not passed. This is required for the agent to work and contains the JWT to authenticate with the server.")
+                if std::env::var("AGENT_TOKEN").is_err()
+                    && std::env::var(crate::agent_workers::AGENT_IDENTITY_TOKEN_FILE_ENV).is_err()
+                {
+                    println!("Neither AGENT_TOKEN nor AGENT_IDENTITY_TOKEN_FILE is passed. One is required for the agent to authenticate with the server.")
                 }
 
                 #[cfg(not(feature = "enterprise"))]

frontend/src/lib/components/InstanceSetting.svelte

@@ -21,6 +21,7 @@
 	import SimpleEditor from './SimpleEditor.svelte'
 	import CriticalAlertChannels from './instanceSettings/CriticalAlertChannels.svelte'
 	import RetentionPeriodOverrides from './instanceSettings/RetentionPeriodOverrides.svelte'
+	import AgentWorkloadIdentityIssuers from './instanceSettings/AgentWorkloadIdentityIssuers.svelte'
 	import SmtpSettings from './instanceSettings/SmtpSettings.svelte'
 	import SecretBackendConfig from './instanceSettings/SecretBackendConfig.svelte'
 	import ExternalInstancePgSettings from './instanceSettings/ExternalInstancePgSettings.svelte'
@@ -629,6 +630,8 @@
 							</Button>
 						</div>
 					</div>
+				{:else if setting.fieldType == 'agent_workload_identity_issuers'}
+					<AgentWorkloadIdentityIssuers {values} disabled={!$enterpriseLicense} />
 				{:else if setting.fieldType == 'critical_error_channels'}
 					<CriticalAlertChannels {values} {openSmtpSettings} {oauths} />
 				{:else if setting.fieldType == 'otel'}

frontend/src/lib/components/InstanceSettings.svelte

@@ -1204,6 +1204,12 @@
 			{#if $enterpriseLicense}
 				<OidcSigningStatus />
 			{/if}
+		{:else if category == 'Agent workers'}
+			<SettingsPageHeader
+				title="Agent workers"
+				description="Optional. Agent workers authenticate with the agent token (AGENT_TOKEN) you create on the Workers page with New agent worker, which needs nothing set here. Add an issuer only to also accept workload identity tokens."
+				link="https://www.windmill.dev/docs/core_concepts/agent_workers"
+			/>
 		{:else if category == 'Auth/OAuth/SAML'}
 			<AuthSettings
 				bind:oauths

frontend/src/lib/components/instanceSettings.ts

@@ -77,6 +77,7 @@ export interface Setting {
 		| 'retention_overrides'
 		| 'instance_banner'
 		| 'accent_color'
+		| 'agent_workload_identity_issuers'
 	storage: SettingStorage
 	advancedToggle?: {
 		label: string
@@ -830,6 +831,17 @@ export const settings: Record<string, Setting[]> = {
 			ee_only: ''
 		}
 	],
+	'Agent workers': [
+		{
+			label: 'Workload identity issuers',
+			key: 'agent_workload_identity_issuers',
+			description:
+				'An issuer lets an agent worker authenticate with a short-lived token its platform issues it (a Kubernetes projected service account token, a SPIFFE JWT-SVID, ...) in place of an agent token: point AGENT_IDENTITY_TOKEN_FILE at it. The token must be signed by a trusted issuer, carry its audience, and match a rule, which decides the worker group and tags the agent may serve. <a href="https://www.windmill.dev/docs/core_concepts/agent_workers">Learn more</a>',
+			fieldType: 'agent_workload_identity_issuers',
+			storage: 'setting',
+			ee_only: ''
+		}
+	],
 	'DB Health': [],
 	'OIDC signing': [],
 	Registries: [
@@ -1338,6 +1350,13 @@ export const instanceSettingsNavigationGroups = [
 				aiDescription: 'Instance SCIM/SAML settings',
 				isEE: true
 			},
+			{
+				id: 'agent_workers',
+				label: 'Agent workers',
+				aiId: 'instance-settings-agent-workers',
+				aiDescription: 'Workload identity issuers agent workers can authenticate with',
+				isEE: true
+			},
 			{
 				id: 'oidc_signing',
 				label: 'OIDC signing',
@@ -1502,6 +1521,7 @@ export const tabToCategoryMap: Record<string, string> = {
 	websocket: 'WebSocket',
 	db_health: 'DB Health',
 	oidc_signing: 'OIDC signing',
+	agent_workers: 'Agent workers',
 	lsp: 'LSP'
 }
 
@@ -1564,6 +1584,7 @@ export const categoryToTabMap: Record<string, string> = {
 	WebSocket: 'websocket',
 	'DB Health': 'db_health',
 	'OIDC signing': 'oidc_signing',
+	'Agent workers': 'agent_workers',
 	LSP: 'lsp'
 }
 

frontend/src/lib/components/instanceSettings/AgentWorkloadIdentityIssuers.svelte

@@ -0,0 +1,276 @@
+<script lang="ts">
+	import type { Writable } from 'svelte/store'
+	import { Button } from '../common'
+	import TextInput from '../text_input/TextInput.svelte'
+	import ToggleButtonGroup from '../common/toggleButton-v2/ToggleButtonGroup.svelte'
+	import ToggleButton from '../common/toggleButton-v2/ToggleButton.svelte'
+	import { Plus, Trash2, X } from 'lucide-svelte'
+
+	interface Props {
+		values: Writable<Record<string, any>>
+		disabled?: boolean
+	}
+
+	let { values, disabled = false }: Props = $props()
+
+	const KEY = 'agent_workload_identity_issuers'
+
+	type KeySource = 'discovery' | 'jwks_uri' | 'inline'
+	type ClaimRow = { id: number; path: string; value: string }
+	type RuleRow = { id: number; claims: ClaimRow[]; worker_group: string; tags: string }
+	type IssuerRow = {
+		id: number
+		issuer: string
+		audience: string
+		keySource: KeySource
+		jwks_uri: string
+		jwks: string
+		rules: RuleRow[]
+	}
+
+	let nextId = 0
+	// Rows keep a stable identity while editing; `lastSynced` tells our own writes to
+	// `$values[KEY]` apart from external ones (load, save, discard), which re-hydrate.
+	let issuers = $state<IssuerRow[]>([])
+	let lastSynced: string | undefined = undefined
+
+	function newRule(): RuleRow {
+		return {
+			id: nextId++,
+			claims: [{ id: nextId++, path: 'sub', value: '' }],
+			worker_group: '',
+			tags: ''
+		}
+	}
+
+	function toRows(v: unknown): IssuerRow[] {
+		if (!Array.isArray(v)) return []
+		return v.map((i: any) => ({
+			id: nextId++,
+			issuer: i.issuer ?? '',
+			audience: i.audience ?? '',
+			keySource: i.jwks ? 'inline' : i.jwks_uri ? 'jwks_uri' : 'discovery',
+			jwks_uri: i.jwks_uri ?? '',
+			jwks: i.jwks ?? '',
+			rules: (i.rules ?? []).map((r: any) => ({
+				id: nextId++,
+				claims: Object.entries(r.claims ?? {}).map(([path, value]) => ({
+					id: nextId++,
+					path,
+					value: String(value)
+				})),
+				worker_group: r.worker_group ?? '',
+				tags: (r.tags ?? []).join(', ')
+			}))
+		}))
+	}
+
+	function serialize(rows: IssuerRow[]): any[] | undefined {
+		if (rows.length === 0) return undefined
+		return rows.map((i) => ({
+			issuer: i.issuer.trim(),
+			...(i.keySource === 'jwks_uri' && i.jwks_uri.trim() ? { jwks_uri: i.jwks_uri.trim() } : {}),
+			...(i.keySource === 'inline' && i.jwks.trim() ? { jwks: i.jwks.trim() } : {}),
+			audience: i.audience.trim(),
+			rules: i.rules.map((r) => ({
+				claims: Object.fromEntries(
+					r.claims.filter((c) => c.path.trim()).map((c) => [c.path.trim(), c.value])
+				),
+				worker_group: r.worker_group.trim(),
+				tags: r.tags
+					.split(',')
+					.map((t) => t.trim())
+					.filter((t) => t)
+			}))
+		}))
+	}
+
+	$effect(() => {
+		const cur = JSON.stringify($values[KEY] ?? null)
+		if (cur === lastSynced) return
+		lastSynced = cur
+		issuers = toRows($values[KEY])
+	})
+
+	$effect(() => {
+		const next = serialize(issuers)
+		if (lastSynced === undefined) return
+		const nextStr = JSON.stringify(next ?? null)
+		// Compare against the stored value normalized the same way, so loading a value whose
+		// keys are ordered differently does not mark the form dirty.
+		if (nextStr === JSON.stringify(serialize(toRows($values[KEY])) ?? null)) return
+		lastSynced = nextStr
+		$values[KEY] = next
+	})
+
+	function addIssuer() {
+		issuers.push({
+			id: nextId++,
+			issuer: '',
+			audience: 'windmill',
+			keySource: 'discovery',
+			jwks_uri: '',
+			jwks: '',
+			rules: [newRule()]
+		})
+	}
+</script>
+
+<div class="flex flex-col gap-4">
+	{#each issuers as issuer, i (issuer.id)}
+		<div class="flex flex-col gap-3 p-4 rounded-md border">
+			<div class="flex items-center justify-between">
+				<span class="text-xs font-semibold text-emphasis">Issuer {i + 1}</span>
+				<Button
+					variant="subtle"
+					unifiedSize="xs"
+					startIcon={{ icon: Trash2 }}
+					iconOnly
+					{disabled}
+					aria-label="Remove issuer"
+					onclick={() => issuers.splice(i, 1)}
+				/>
+			</div>
+			<label class="flex flex-col gap-1">
+				<span class="text-xs text-secondary">Issuer (must equal the token's iss)</span>
+				<TextInput
+					bind:value={issuer.issuer}
+					inputProps={{ placeholder: 'https://kubernetes.default.svc.cluster.local', disabled }}
+				/>
+			</label>
+			<label class="flex flex-col gap-1">
+				<span class="text-xs text-secondary">Audience (must be in the token's aud)</span>
+				<TextInput
+					bind:value={issuer.audience}
+					inputProps={{ placeholder: 'windmill', disabled }}
+				/>
+			</label>
+			<div class="flex flex-col gap-1">
+				<span class="text-xs text-secondary">Signing keys</span>
+				<ToggleButtonGroup bind:selected={issuer.keySource} {disabled}>
+					{#snippet children({ item })}
+						<ToggleButton
+							value="discovery"
+							label="OIDC discovery"
+							tooltip="Read jwks_uri from <issuer>/.well-known/openid-configuration"
+							{item}
+						/>
+						<ToggleButton value="jwks_uri" label="JWKS URL" {item} />
+						<ToggleButton
+							value="inline"
+							label="Inline JWKS"
+							tooltip="For an issuer this instance cannot reach, e.g. the output of kubectl get --raw /openid/v1/jwks"
+							{item}
+						/>
+					{/snippet}
+				</ToggleButtonGroup>
+				{#if issuer.keySource === 'jwks_uri'}
+					<TextInput
+						bind:value={issuer.jwks_uri}
+						inputProps={{ placeholder: 'https://issuer.example.com/keys', disabled }}
+					/>
+				{:else if issuer.keySource === 'inline'}
+					<TextInput
+						underlyingInputEl="textarea"
+						bind:value={issuer.jwks}
+						inputProps={{ placeholder: '{"keys": [...]}', rows: 4, disabled }}
+					/>
+				{/if}
+			</div>
+			<div class="flex flex-col gap-2">
+				<span class="text-xs text-secondary">
+					Rules, tried in order: the first whose claims all match gives the agent its worker group
+					and tags. A token no rule matches is refused.
+				</span>
+				{#each issuer.rules as rule, j (rule.id)}
+					<div class="flex flex-col gap-2 border-l-2 border-surface-selected pl-3">
+						<div class="flex items-center justify-between">
+							<span class="text-2xs font-semibold text-secondary">Rule {j + 1}</span>
+							<Button
+								variant="subtle"
+								unifiedSize="xs"
+								startIcon={{ icon: Trash2 }}
+								iconOnly
+								{disabled}
+								aria-label="Remove rule"
+								onclick={() => issuer.rules.splice(j, 1)}
+							/>
+						</div>
+						{#each rule.claims as claim, k (claim.id)}
+							<div class="flex items-center gap-2">
+								<TextInput
+									bind:value={claim.path}
+									class="w-64"
+									inputProps={{ placeholder: 'kubernetes.io.namespace', disabled }}
+								/>
+								<span class="text-xs text-secondary">=</span>
+								<TextInput
+									bind:value={claim.value}
+									inputProps={{ placeholder: 'system:serviceaccount:<ns>:<sa>', disabled }}
+								/>
+								<Button
+									variant="subtle"
+									unifiedSize="xs"
+									startIcon={{ icon: X }}
+									iconOnly
+									disabled={disabled || rule.claims.length <= 1}
+									aria-label="Remove claim"
+									onclick={() => rule.claims.splice(k, 1)}
+								/>
+							</div>
+						{/each}
+						<div class="flex">
+							<Button
+								variant="subtle"
+								unifiedSize="xs"
+								startIcon={{ icon: Plus }}
+								{disabled}
+								onclick={() => rule.claims.push({ id: nextId++, path: '', value: '' })}
+							>
+								Add claim
+							</Button>
+						</div>
+						<div class="flex gap-2">
+							<label class="flex flex-col gap-1 w-48">
+								<span class="text-2xs text-secondary">Worker group</span>
+								<TextInput
+									bind:value={rule.worker_group}
+									inputProps={{ placeholder: 'k8s', disabled }}
+								/>
+							</label>
+							<label class="flex flex-col gap-1 grow">
+								<span class="text-2xs text-secondary">Tags (comma separated)</span>
+								<TextInput
+									bind:value={rule.tags}
+									inputProps={{ placeholder: 'dbt, etl', disabled }}
+								/>
+							</label>
+						</div>
+					</div>
+				{/each}
+				<div class="flex">
+					<Button
+						variant="default"
+						unifiedSize="sm"
+						startIcon={{ icon: Plus }}
+						{disabled}
+						onclick={() => issuer.rules.push(newRule())}
+					>
+						Add rule
+					</Button>
+				</div>
+			</div>
+		</div>
+	{/each}
+	<div class="flex">
+		<Button
+			variant="default"
+			unifiedSize="sm"
+			startIcon={{ icon: Plus }}
+			{disabled}
+			onclick={addIssuer}
+		>
+			Add issuer
+		</Button>
+	</div>
+</div>