kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: require AGENT_TOKEN and BASE_INTERNAL_URL for agent mode

Browse Source 
- Add AgentConfig struct to validate required env vars on startup
- Change build_agent_http_client to require explicit token and URL
- Remove DEFAULT_BASE_INTERNAL_URL fallback (no more silent localhost:8000)
- Exit immediately if agent cannot connect to server on initial load
- Update integration tests to use dynamic port for BASE_INTERNAL_URL

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-01-29 23:12:07 +00:00
parent 40a883a987
commit e3cccf97b0
5 changed files with 169 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
backend/src/main.rs
View File

@@ -34,7 +34,7 @@ use windmill_common::ee_oss::{
};
use windmill_common::{
agent_workers::build_agent_http_client,
agent_workers::AgentConfig,
global_settings::{
APP_WORKSPACED_ROUTE_SETTING, BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING,
CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
@@ -43,13 +43,14 @@ use windmill_common::{
EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING,
HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, INDEXER_SETTING,
INSTANCE_PYTHON_VERSION_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING,
KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MAVEN_REPOS_SETTING, OTEL_TRACING_PROXY_SETTING,
KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MAVEN_REPOS_SETTING,
MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NO_DEFAULT_MAVEN_SETTING,
NPM_CONFIG_REGISTRY_SETTING, NUGET_CONFIG_SETTING, OAUTH_SETTING, OTEL_SETTING,
PIP_INDEX_URL_SETTING, POWERSHELL_REPO_PAT_SETTING, POWERSHELL_REPO_URL_SETTING,
REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
RETENTION_PERIOD_SECS_SETTING, RUBY_REPOS_SETTING, SAML_METADATA_SETTING,
SCIM_TOKEN_SETTING, SMTP_SETTING, TEAMS_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
OTEL_TRACING_PROXY_SETTING, PIP_INDEX_URL_SETTING, POWERSHELL_REPO_PAT_SETTING,
POWERSHELL_REPO_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
RUBY_REPOS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING, TEAMS_SETTING,
TIMEOUT_WAIT_RESULT_SETTING,
},
scripts::ScriptLang,
stats_oss::schedule_stats,
@@ -99,9 +100,10 @@ use crate::monitor::{
reload_bunfig_install_scopes_setting, reload_critical_alert_mute_ui_setting,
reload_critical_error_channels_setting, reload_extra_pip_index_url_setting,
reload_hub_api_secret_setting, reload_hub_base_url_setting, reload_job_default_timeout_setting,
reload_jwt_secret_setting, reload_license_key, reload_otel_tracing_proxy_setting,
reload_npm_config_registry_setting, reload_pip_index_url_setting, reload_retention_period_setting,
reload_scim_token_setting, reload_smtp_config, reload_worker_config, MonitorIteration,
reload_jwt_secret_setting, reload_license_key, reload_npm_config_registry_setting,
reload_otel_tracing_proxy_setting, reload_pip_index_url_setting,
reload_retention_period_setting, reload_scim_token_setting, reload_smtp_config,
reload_worker_config, MonitorIteration,
};
#[cfg(feature = "parquet")]
@@ -410,7 +412,10 @@ pub async fn sync_cached_resource_types(db: &sqlx::Pool<sqlx::Postgres>) -> anyh
let cache_path = format!("{}/{}", HUB_RT_CACHE_DIR, HUB_RT_CACHE_FILE);
if tokio::fs::metadata(&cache_path).await.is_err() {
tracing::info!("No cached resource types found at {}, skipping sync", cache_path);
tracing::info!(
"No cached resource types found at {}, skipping sync",
cache_path
);
return Ok(());
}
@@ -420,8 +425,8 @@ pub async fn sync_cached_resource_types(db: &sqlx::Pool<sqlx::Postgres>) -> anyh
.await
.with_context(|| format!("Failed to read cache file from {}", cache_path))?;
let cached_types: Vec<HubResourceType> = serde_json::from_str(&content)
.with_context(|| "Failed to parse cached resource types")?;
let cached_types: Vec<HubResourceType> =
serde_json::from_str(&content).with_context(|| "Failed to parse cached resource types")?;
tracing::info!("Found {} cached resource types", cached_types.len());
@@ -433,11 +438,13 @@ pub async fn sync_cached_resource_types(db: &sqlx::Pool<sqlx::Postgres>) -> anyh
.await
.with_context(|| "Failed to fetch existing resource types")?;
let existing_map: std::collections::HashMap<String, (Option<serde_json::Value>, Option<String>)> =
existing_types
.into_iter()
.map(|(name, schema, desc)| (name, (schema, desc)))
.collect();
let existing_map: std::collections::HashMap<
String,
(Option<serde_json::Value>, Option<String>),
> = existing_types
.into_iter()
.map(|(name, schema, desc)| (name, (schema, desc)))
.collect();
let mut synced_count = 0;
let mut skipped_count = 0;
@@ -478,36 +485,45 @@ pub async fn sync_cached_resource_types(db: &sqlx::Pool<sqlx::Postgres>) -> anyh
}
fn print_help() {
println!("Windmill - a fast, open-source workflow engine and job runner.");
println!();
println!("Usage:");
println!(" windmill [SUBCOMMAND]");
println!();
println!("Subcommands:");
println!(" help | -h | --help Show this help information and exit");
println!(" version Show Windmill version and exit");
println!(" cache [hubPaths.json] Pre-cache hub scripts (default: ./hubPaths.json)");
println!(" cache-rt Pre-cache hub resource types");
println!();
println!("Environment variables (name = default):");
println!(" DATABASE_URL = <required> The Postgres database url.");
println!(" MODE = standalone Mode: standalone | worker | server | agent");
println!(" BASE_URL = http://localhost:8000 Public base URL of your instance (overridden by instance settings)");
println!(" PORT = {} HTTP port (server/indexer/MCP modes)", DEFAULT_PORT);
println!(" SERVER_BIND_ADDR = <mode dependent> IP to bind to (server: {}, worker: {})", DEFAULT_SERVER_BIND_ADDR, DEFAULT_WORKER_BIND_ADDR);
println!(" NUM_WORKERS = {} Number of workers (standalone/worker modes)", DEFAULT_NUM_WORKERS);
println!(" WORKER_GROUP = default Worker group this worker belongs to",);
println!(" JSON_FMT = false Output logs in JSON instead of logfmt");
println!(" METRICS_ADDR = None (EE only) Prometheus metrics addr at /metrics; set \"true\" to use :8001");
println!(" SUPERADMIN_SECRET = None Virtual superadmin token (server)");
println!(" LICENSE_KEY = None (EE only) Enterprise license key (workers require valid key)");
println!(" RUN_UPDATE_CA_CERTIFICATE_AT_START = false Run system CA update at startup");
println!(" RUN_UPDATE_CA_CERTIFICATE_PATH = /usr/sbin/update-ca-certificates Path to CA update tool");
println!(" SYNC_CACHED_RT = false Sync cached resource types to admins workspace on server start");
println!();
println!("Notes:");
println!("- Advanced and less commonly used settings are managed via the database and are omitted here.");
println!("- At startup, Windmill logs currently set configuration keys for visibility.");
println!("Windmill - a fast, open-source workflow engine and job runner.");
println!();
println!("Usage:");
println!(" windmill [SUBCOMMAND]");
println!();
println!("Subcommands:");
println!(" help | -h | --help Show this help information and exit");
println!(" version Show Windmill version and exit");
println!(" cache [hubPaths.json] Pre-cache hub scripts (default: ./hubPaths.json)");
println!(" cache-rt Pre-cache hub resource types");
println!();
println!("Environment variables (name = default):");
println!(" DATABASE_URL = <required> The Postgres database url.");
println!(" MODE = standalone Mode: standalone | worker | server | agent");
println!(" BASE_URL = http://localhost:8000 Public base URL of your instance (overridden by instance settings)");
println!(
" PORT = {} HTTP port (server/indexer/MCP modes)",
DEFAULT_PORT
);
println!(
" SERVER_BIND_ADDR = <mode dependent> IP to bind to (server: {}, worker: {})",
DEFAULT_SERVER_BIND_ADDR, DEFAULT_WORKER_BIND_ADDR
);
println!(
" NUM_WORKERS = {} Number of workers (standalone/worker modes)",
DEFAULT_NUM_WORKERS
);
println!(" WORKER_GROUP = default Worker group this worker belongs to",);
println!(" JSON_FMT = false Output logs in JSON instead of logfmt");
println!(" METRICS_ADDR = None (EE only) Prometheus metrics addr at /metrics; set \"true\" to use :8001");
println!(" SUPERADMIN_SECRET = None Virtual superadmin token (server)");
println!(" LICENSE_KEY = None (EE only) Enterprise license key (workers require valid key)");
println!(" RUN_UPDATE_CA_CERTIFICATE_AT_START = false Run system CA update at startup");
println!(" RUN_UPDATE_CA_CERTIFICATE_PATH = /usr/sbin/update-ca-certificates Path to CA update tool");
println!(" SYNC_CACHED_RT = false Sync cached resource types to admins workspace on server start");
println!();
println!("Notes:");
println!("- Advanced and less commonly used settings are managed via the database and are omitted here.");
println!("- At startup, Windmill logs currently set configuration keys for visibility.");
}
async fn windmill_main() -> anyhow::Result<()> {
@@ -641,15 +657,23 @@ async fn windmill_main() -> anyhow::Result<()> {
.and_then(|x| x.parse().ok())
.unwrap_or(IpAddr::from(default_bind_addr));
let (conn, first_suffix) = if mode == Mode::Agent {
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);
}
};
tracing::info!(
"Creating http client for cluster using base internal url {}",
std::env::var("BASE_INTERNAL_URL").unwrap_or_default()
agent_config.base_internal_url
);
let suffix = create_default_worker_suffix(&hostname);
(
Connection::Http(build_agent_http_client(&suffix, None, None)),
Connection::Http(agent_config.build_http_client(&suffix)),
Some(suffix),
Some(agent_config),
)
} else {
println!("Connecting to database...");
@@ -673,7 +697,8 @@ async fn windmill_main() -> anyhow::Result<()> {
reload_otel_tracing_proxy_setting(&Connection::Sql(db.clone())).await;
#[cfg(feature = "deno_core")]
if windmill_worker::is_otel_tracing_proxy_enabled_for_lang(&ScriptLang::Nativets).await {
if windmill_worker::is_otel_tracing_proxy_enabled_for_lang(&ScriptLang::Nativets).await
{
match windmill_worker::load_internal_otel_exporter().await {
Ok(()) => {
tracing::info!("Internal OTEL exporter initialized for nativets tracing");
@@ -688,7 +713,7 @@ async fn windmill_main() -> anyhow::Result<()> {
load_otel(&db).await;
println!("Database connected");
(Connection::Sql(db), None)
(Connection::Sql(db), None, None)
};
let environment = if let Ok(environment) = std::env::var("OTEL_ENVIRONMENT") {
@@ -799,6 +824,12 @@ Windmill Community Edition {GIT_VERSION}
// if key still invalid and num_workers > 0, set to 0
if let Err(err) = reload_license_key(&conn).await {
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"
);
std::process::exit(1);
}
}
let valid_key = *LICENSE_KEY_VALID.read().await;
if !valid_key && !server_mode {
@@ -1057,7 +1088,12 @@ Windmill Community Edition {GIT_VERSION}
conn: if i == 0 || mode != Mode::Agent {
conn.clone()
} else {
Connection::Http(build_agent_http_client(&suffix, None, None))
Connection::Http(
agent_config
.as_ref()
.expect("agent_config must be set in agent mode")
.build_http_client(&suffix),
)
},
worker_name: worker_name_with_suffix(
mode == Mode::Agent,

32
backend/tests/common/mod.rs
View File

@@ -764,23 +764,25 @@ pub async fn run_preview_relative_imports(
#[cfg(all(feature = "private", feature = "agent_worker_server"))]
pub async fn testing_http_connection(port: u16) -> Connection {
let suffix = windmill_common::utils::create_default_worker_suffix("test-agent-worker");
let agent_token = format!(
"{}{}",
windmill_common::agent_workers::AGENT_JWT_PREFIX,
windmill_common::jwt::encode_with_internal_secret(
windmill_api::agent_workers_ee::AgentAuth {
worker_group: "testing-agent".to_owned(),
suffix: Some(suffix.clone()),
tags: vec!["flow".into(), "python3".into(), "dependency".into()],
exp: Some(usize::MAX),
}
)
.await
.expect("JWT token to be created")
);
let base_internal_url = format!("http://localhost:{port}");
Connection::Http(windmill_common::agent_workers::build_agent_http_client(
&suffix,
Some(format!(
"{}{}",
windmill_common::agent_workers::AGENT_JWT_PREFIX,
windmill_common::jwt::encode_with_internal_secret(
windmill_api::agent_workers_ee::AgentAuth {
worker_group: "testing-agent".to_owned(),
suffix: Some(suffix.clone()),
tags: vec!["flow".into(), "python3".into(), "dependency".into()],
exp: Some(usize::MAX),
}
)
.await
.expect("JWT token to be created")
)),
Some(format!("http://localhost:{port}")),
&agent_token,
&base_internal_url,
))
}

63
backend/windmill-common/src/agent_workers.rs
View File

@@ -13,14 +13,55 @@ 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.
#[derive(Clone)]
pub struct AgentConfig {
pub agent_token: String,
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 base_internal_url = std::env::var("BASE_INTERNAL_URL")
.map_err(|_| AgentConfigError::MissingBaseInternalUrl)?;
Ok(Self { agent_token, 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)
}
}
#[derive(Debug)]
pub enum AgentConfigError {
MissingAgentToken,
MissingBaseInternalUrl,
}
impl std::fmt::Display for AgentConfigError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AgentConfigError::MissingAgentToken => {
write!(f, "AGENT_TOKEN environment variable is not set but required for agent mode")
}
AgentConfigError::MissingBaseInternalUrl => {
write!(f, "BASE_INTERNAL_URL environment variable is not set but required for agent mode")
}
}
}
}
impl std::error::Error for AgentConfigError {}
lazy_static! {
pub static ref AGENT_TOKEN: String = std::env::var("AGENT_TOKEN").unwrap_or_default();
pub static ref DECODED_AGENT_TOKEN: Option<AgentAuth> = {
if AGENT_TOKEN.is_empty() {
None
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 {
decode_without_verify::<AgentAuth>(AGENT_TOKEN.trim_start_matches(AGENT_JWT_PREFIX))
.ok()
None
}
};
}
@@ -28,7 +69,7 @@ lazy_static! {
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct AgentAuth {
pub worker_group: String,
pub suffix: Option<String>,
pub suffix: Option<String>,
pub tags: Vec<String>,
pub exp: Option<usize>,
}
@@ -37,8 +78,8 @@ pub const AGENT_JWT_PREFIX: &str = "jwt_agent_";
pub fn build_agent_http_client(
worker_suffix: &str,
agent_token: Option<String>,
base_internal_url: Option<String>,
agent_token: &str,
base_internal_url: &str,
) -> HttpClient {
let client = ClientBuilder::new(
configure_client(
@@ -58,9 +99,7 @@ pub fn build_agent_http_client(
"{}{}_{}",
AGENT_JWT_PREFIX,
worker_suffix,
agent_token
.unwrap_or(AGENT_TOKEN.clone())
.trim_start_matches(AGENT_JWT_PREFIX)
agent_token.trim_start_matches(AGENT_JWT_PREFIX)
);
headers.insert(
"Authorization",
@@ -76,7 +115,7 @@ pub fn build_agent_http_client(
))
.build();
HttpClient { client, base_internal_url }
HttpClient { client, base_internal_url: base_internal_url.to_string() }
}
#[derive(Deserialize, Serialize)]

15
backend/windmill-common/src/utils.rs
View File

@@ -151,10 +151,6 @@ lazy_static::lazy_static! {
pub static ref HUB_API_SECRET: Arc<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
}
lazy_static::lazy_static! {
pub static ref AGENT_TOKEN: String = std::env::var("AGENT_TOKEN").unwrap_or_default();
}
#[derive(Clone)]
pub struct ModeAndAddons {
pub indexer: bool,
@@ -1190,7 +1186,8 @@ mod tests {
#[test]
fn test_merge_nested_raw_values_to_array_complex_types() {
let val1 = serde_json::value::RawValue::from_string("{\"name\":\"Alice\"}".to_string()).unwrap();
let val1 =
serde_json::value::RawValue::from_string("{\"name\":\"Alice\"}".to_string()).unwrap();
let val2 = serde_json::value::RawValue::from_string("[1,2,3]".to_string()).unwrap();
let val3 = serde_json::value::RawValue::from_string("\"text\"".to_string()).unwrap();
let val4 = serde_json::value::RawValue::from_string("null".to_string()).unwrap();
@@ -1236,7 +1233,13 @@ mod tests {
let inner3 = vec![val3];
let inner4 = vec![val4];
let inner5 = vec![val5];
let nested = vec![inner1.iter(), inner2.iter(), inner3.iter(), inner4.iter(), inner5.iter()];
let nested = vec![
inner1.iter(),
inner2.iter(),
inner3.iter(),
inner4.iter(),
inner5.iter(),
];
let result = merge_nested_raw_values_to_array(nested.into_iter());

16
backend/windmill-common/src/worker.rs
View File

@@ -274,13 +274,10 @@ lazy_static::lazy_static! {
pub const ROOT_CACHE_NOMOUNT_DIR: &str = concatcp!(TMP_DIR, "/cache_nomount/");
pub static MIN_VERSION_IS_LATEST: AtomicBool = AtomicBool::new(false);
const DEFAULT_BASE_INTERNAL_URL: &str = "http://localhost:8000";
#[derive(Clone)]
pub struct HttpClient {
pub client: ClientWithMiddleware,
pub base_internal_url: Option<String>,
pub base_internal_url: String,
}
impl Deref for HttpClient {
@@ -298,10 +295,7 @@ impl HttpClient {
headers: Option<HeaderMap>,
body: &T,
) -> anyhow::Result<R> {
let base_url = self
.base_internal_url
.clone()
.unwrap_or(DEFAULT_BASE_INTERNAL_URL.to_owned());
let base_url = self.base_internal_url.clone();
let response_builder = self.client.post(format!("{}{}", base_url, url)).json(body);
@@ -327,11 +321,7 @@ impl HttpClient {
}
pub async fn get<R: DeserializeOwned>(&self, url: &str) -> anyhow::Result<R> {
let base_url = self
.base_internal_url
.clone()
.unwrap_or(DEFAULT_BASE_INTERNAL_URL.to_owned());
let base_url = self.base_internal_url.clone();
let response = self
.client
.get(format!("{}{}", base_url, url))