diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 567af4bbcc..9d890d9b05 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15225,6 +15225,7 @@ dependencies = [ "axum 0.8.9", "chrono", "futures", + "hex", "http 1.4.1", "hyper 1.10.1", "lazy_static", @@ -15232,6 +15233,7 @@ dependencies = [ "reqwest 0.13.1", "serde", "serde_json", + "sha2 0.10.9", "sql-builder", "sqlx", "tokio", diff --git a/backend/windmill-api-integration-tests/tests/fixtures/resource_cache_rls.sql b/backend/windmill-api-integration-tests/tests/fixtures/resource_cache_rls.sql new file mode 100644 index 0000000000..8c30eab5b2 --- /dev/null +++ b/backend/windmill-api-integration-tests/tests/fixtures/resource_cache_rls.sql @@ -0,0 +1,23 @@ +-- Fixture for the resource-value interpolation cache RLS regression test. +-- Extends base.sql (which defines test-user [admin], test-user-2, test-user-3 +-- and their tokens). +-- +-- A folder `secret` is readable ONLY by test-user-2 (via extra_perms). It holds a +-- variable and a resource that interpolates it. test-user-3 has no access to the +-- folder, so a cache entry warmed by test-user-2 with allow_cache=true must never +-- be served back to test-user-3. + +INSERT INTO folder (workspace_id, name, display_name, owners, extra_perms, created_by) +VALUES ('test-workspace', 'secret', 'Secret Folder', '{}', + '{"u/test-user-2": true}', 'test-user'); + +-- A (non-secret) variable gated to the `secret` folder; its value gets interpolated +-- into the resource value below and ends up in the cached, already-resolved blob. +INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms) +VALUES ('test-workspace', 'f/secret/db_password', 'LEAKED_FOLDER_SECRET', false, + 'Folder-gated secret', '{}'); + +INSERT INTO resource (workspace_id, path, value, description, resource_type, extra_perms, created_by) +VALUES ('test-workspace', 'f/secret/cache_target', + '{"host": "db.internal", "password": "$var:f/secret/db_password"}', + 'Folder-gated resource referencing a folder-gated variable', 'object', '{}', 'test-user'); diff --git a/backend/windmill-api-integration-tests/tests/fixtures/variable_cache_rls.sql b/backend/windmill-api-integration-tests/tests/fixtures/variable_cache_rls.sql new file mode 100644 index 0000000000..69a6810b7b --- /dev/null +++ b/backend/windmill-api-integration-tests/tests/fixtures/variable_cache_rls.sql @@ -0,0 +1,15 @@ +-- Fixture for the variable-value cache RLS regression test. +-- Extends base.sql (which defines test-user [admin], test-user-2, test-user-3 +-- and their tokens). +-- +-- A folder `secret` is readable ONLY by test-user-2 (via extra_perms). It holds a +-- variable that test-user-2 can read but test-user-3 cannot. A cache entry warmed +-- by test-user-2 with allow_cache=true must never be served back to test-user-3. + +INSERT INTO folder (workspace_id, name, display_name, owners, extra_perms, created_by) +VALUES ('test-workspace', 'secret', 'Secret Folder', '{}', + '{"u/test-user-2": true}', 'test-user'); + +INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms) +VALUES ('test-workspace', 'f/secret/cache_target_var', 'LEAKED_VAR_SECRET', false, + 'Folder-gated variable', '{}'); diff --git a/backend/windmill-api-integration-tests/tests/resources.rs b/backend/windmill-api-integration-tests/tests/resources.rs index 363217712f..cc78056176 100644 --- a/backend/windmill-api-integration-tests/tests/resources.rs +++ b/backend/windmill-api-integration-tests/tests/resources.rs @@ -477,6 +477,117 @@ async fn test_resource_endpoints(db: Pool) -> anyhow::Result<()> { Ok(()) } +/// Regression test: the resource-value interpolation cache +/// (`get_value_interpolated?allow_cache=true`) must be identity-scoped. test-user-2 +/// (folder access) warms the cache; test-user-3 (no access) must then be denied rather +/// than served the cached, already-decrypted value. Pre-fix the unscoped key returned +/// a 200 with the secret here. +#[sqlx::test(migrations = "../migrations", fixtures("base", "resource_cache_rls"))] +async fn test_resource_value_cache_is_identity_scoped(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let url = format!( + "{}?allow_cache=true", + resource_url(port, "get_value_interpolated", "f/secret/cache_target") + ); + let get = |token: &str| { + client() + .get(url.as_str()) + .header("Authorization", format!("Bearer {token}")) + }; + + // test-user-2 has folder access and WARMS the cache. + let resp = get("SECRET_TOKEN_2").send().await?; + assert_eq!(resp.status(), 200); + assert!(resp.text().await?.contains("LEAKED_FOLDER_SECRET")); + + // test-user-3 has no folder access: must miss the cache and be denied (401), not leak. + let resp = get("SECRET_TOKEN_3").send().await?; + assert_eq!(resp.status(), 401); + assert!(!resp.text().await?.contains("LEAKED_FOLDER_SECRET")); + + Ok(()) +} + +/// A resource whose value contains a `$WM_*` contextual variable (e.g. `$WM_TOKEN`) is +/// job-dependent and must NEVER be cached — even when first read WITHOUT a `job_id`, where the +/// placeholder is left unresolved (caching that would serve a stale placeholder to a later job +/// read). Any other value — plain, or a non-`$WM_` `$`-string like `$HOME` (which is NOT +/// interpolated, so it's constant) — is job-independent and IS cached, with the entry shared +/// across job contexts (a read carrying a `job_id` still hits it, keeping the hit ratio up). +/// We prove all three by warming each (no job_id), deleting the row directly (cache survives), +/// then re-reading: the job-independent ones are still served from cache — even under a +/// `job_id` — while the `$WM_*` one was never cached and 404s. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_resource_cache_handles_job_context(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let base = format!("http://localhost:{port}/api/w/test-workspace/resources"); + + let plain = "u/test-user/plain_res"; + let dollar = "u/test-user/dollar_res"; // non-$WM_ `$`-string: not interpolated, cacheable + let jobctx = "u/test-user/jobctx_res"; + for (path, value) in [ + (plain, json!({"v": 1})), + (dollar, json!({"d": "$HOME"})), + (jobctx, json!({"j": "$WM_JOB_ID"})), + ] { + let resp = authed(client().post(format!("{base}/create"))) + .json( + &json!({ "path": path, "value": value, "description": "", "resource_type": "object" }), + ) + .send() + .await?; + assert_eq!(resp.status(), 201); + } + + let get = |path: &str, query: &str| { + let url = format!("{base}/get_value_interpolated/{path}?{query}"); + async move { authed(client().get(url)).send().await.unwrap() } + }; + + // Warm all three WITHOUT a job context (the placeholder is left unresolved for `jobctx`). + for path in [plain, dollar, jobctx] { + assert_eq!(get(path, "allow_cache=true").await.status(), 200); + } + + // Delete the rows directly — bypasses the API/NOTIFY, so the in-memory cache survives. + for path in [plain, dollar, jobctx] { + sqlx::query("DELETE FROM resource WHERE workspace_id = 'test-workspace' AND path = $1") + .bind(path) + .execute(&db) + .await?; + } + + // Job-independent values are cached and still served even under a job_id (a random uuid is + // fine: a cache hit short-circuits before any job lookup). `$HOME` is a non-`$WM_` string, + // so it's not interpolated and stays cacheable. + for path in [plain, dollar] { + let resp = get( + path, + "allow_cache=true&job_id=11111111-1111-4111-8111-111111111111", + ) + .await; + assert_eq!( + resp.status(), + 200, + "job-independent resource ({path}) must stay cached and be served under a job_id" + ); + } + + // The `$WM_*` resource was never cached → the (now deleted) row is not found. + let resp = get(jobctx, "allow_cache=true").await; + assert_ne!( + resp.status(), + 200, + "resource with a $WM_* contextual variable must not be cached" + ); + + Ok(()) +} + #[cfg(feature = "mcp")] #[sqlx::test(migrations = "../migrations", fixtures("base", "resources_test"))] async fn test_mcp_tools(db: Pool) -> anyhow::Result<()> { diff --git a/backend/windmill-api-integration-tests/tests/variables.rs b/backend/windmill-api-integration-tests/tests/variables.rs index 0d4edaff91..e5018f4f97 100644 --- a/backend/windmill-api-integration-tests/tests/variables.rs +++ b/backend/windmill-api-integration-tests/tests/variables.rs @@ -108,12 +108,10 @@ async fn test_variable_endpoints(db: Pool) -> anyhow::Result<()> { assert_eq!(secret["value"], serde_json::Value::Null); // list with path_start filter - let resp = authed(client().get(format!( - "{base}/list?path_start=u/test-user/plain" - ))) - .send() - .await - .unwrap(); + let resp = authed(client().get(format!("{base}/list?path_start=u/test-user/plain"))) + .send() + .await + .unwrap(); assert_eq!(resp.status(), 200); let list = resp.json::>().await?; assert_eq!(list.len(), 1); @@ -252,3 +250,91 @@ async fn test_variable_endpoints(db: Pool) -> anyhow::Result<()> { Ok(()) } + +/// Regression test: the variable-value cache (`get_value?allow_cache=true`) must be +/// identity-scoped. test-user-2 (folder access) warms the cache; test-user-3 (no access) +/// must then be denied rather than served the cached value. +#[sqlx::test(migrations = "../migrations", fixtures("base", "variable_cache_rls"))] +async fn test_variable_value_cache_is_identity_scoped(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let url = format!( + "{}?allow_cache=true", + variable_url(port, "get_value", "f/secret/cache_target_var") + ); + let get = |token: &str| { + client() + .get(url.as_str()) + .header("Authorization", format!("Bearer {token}")) + }; + + // test-user-2 has folder access and WARMS the cache. + let resp = get("SECRET_TOKEN_2").send().await?; + assert_eq!(resp.status(), 200); + assert!(resp.text().await?.contains("LEAKED_VAR_SECRET")); + + // test-user-3 has no folder access: must miss the cache and be denied (401), not leak. + let resp = get("SECRET_TOKEN_3").send().await?; + assert_eq!(resp.status(), 401); + assert!(!resp.text().await?.contains("LEAKED_VAR_SECRET")); + + Ok(()) +} + +/// Secret variables ARE cached (with their per-read side effects — the EE +/// `variables.decrypt_secret` audit and running-job secret registration — re-run on every +/// hit; that re-emission is not observable in the OSS build since `audit_log` is a no-op). +/// We assert the caching itself: warm the cache, delete the row directly (no API/NOTIFY, so +/// the in-memory cache survives), and re-read with `allow_cache=true` — the value is still +/// returned from cache. A non-secret variable behaves identically (control). +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_variables_are_cached(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + let base = format!("http://localhost:{port}/api/w/test-workspace/variables"); + + let plain = "u/test-user/cache_plain_probe"; + let secret = "u/test-user/cache_secret_probe"; + + // Create one non-secret and one secret variable (the secret is stored encrypted). + for (path, value, is_secret) in [ + (plain, "PLAIN_PROBE", false), + (secret, "SECRET_PROBE", true), + ] { + let resp = authed(client().post(format!("{base}/create"))) + .json( + &json!({ "path": path, "value": value, "is_secret": is_secret, "description": "" }), + ) + .send() + .await?; + assert_eq!(resp.status(), 201); + } + + let read = |path: &str| { + let url = format!("{base}/get_value/{path}?allow_cache=true"); + async move { authed(client().get(url)).send().await.unwrap() } + }; + + // Warm the cache for both. + assert_eq!(read(plain).await.json::().await?, "PLAIN_PROBE"); + assert_eq!(read(secret).await.json::().await?, "SECRET_PROBE"); + + // Delete both rows directly — bypasses the API and its NOTIFY-based invalidation, so + // the in-memory cache survives. A subsequent read can only succeed from cache. + for path in [plain, secret] { + sqlx::query("DELETE FROM variable WHERE workspace_id = 'test-workspace' AND path = $1") + .bind(path) + .execute(&db) + .await?; + } + + // Both (secret included) are still served from the cache. + assert_eq!(read(plain).await.json::().await?, "PLAIN_PROBE"); + let resp = read(secret).await; + assert_eq!(resp.status(), 200, "secret must still be served from cache"); + assert_eq!(resp.json::().await?, "SECRET_PROBE"); + + Ok(()) +} diff --git a/backend/windmill-store/Cargo.toml b/backend/windmill-store/Cargo.toml index cb1e41ad05..b3aca5e669 100644 --- a/backend/windmill-store/Cargo.toml +++ b/backend/windmill-store/Cargo.toml @@ -45,6 +45,8 @@ tracing.workspace = true uuid.workspace = true quick_cache.workspace = true lazy_static.workspace = true +sha2.workspace = true +hex.workspace = true sql-builder.workspace = true async-recursion.workspace = true futures.workspace = true diff --git a/backend/windmill-store/src/resources.rs b/backend/windmill-store/src/resources.rs index 3e5292e805..402628e3ad 100644 --- a/backend/windmill-store/src/resources.rs +++ b/backend/windmill-store/src/resources.rs @@ -17,7 +17,7 @@ use windmill_common::db::DB; use windmill_common::workspaces::{check_deploy_rules, RuleCheckResult}; use crate::secret_backend_ext::rename_vault_secret; -use crate::var_resource_cache::{cache_resource, get_cached_resource}; +use crate::var_resource_cache::{auth_identity, cache_resource, get_cached_resource}; use windmill_common::utils::{escape_ilike_pattern, BulkDeleteRequest}; use windmill_common::webhook::{WebhookMessage, WebhookShared}; @@ -550,8 +550,18 @@ pub async fn get_resource_value_interpolated_internal<'a>( return Ok(Some(pg_creds)); } - if allow_cache { - if let Some(cached_value) = get_cached_resource(&workspace, &path) { + // Scope the cache to the caller's full authorization identity (not just email): the + // cached value is already decrypted/interpolated under this caller's RLS context, so it + // must never be served to a context that resolves to different permissions. Only + // job-independent values are ever stored (see the write below), so a hit is always safe + // to return regardless of the current `job_id`. + let cache_identity = allow_cache.then(|| match db_with_opt_authed.authed() { + Some(authed) => auth_identity(authed), + None => format!("\0system:{}", db_with_opt_authed.email()), + }); + + if let Some(identity) = cache_identity.as_deref() { + if let Some(cached_value) = get_cached_resource(&workspace, &path, identity) { return Ok(Some(cached_value)); } } @@ -575,17 +585,24 @@ pub async fn get_resource_value_interpolated_internal<'a>( let value = not_found_if_none(value_o, "Resource", path)?; if let Some(value) = value { - let r = transform_json_value( + // Track whether interpolation pulled in a `$WM_*` contextual variable. If it did, the + // result is job-dependent (and may embed `$WM_TOKEN`) and must not be cached; if not, + // it's job-independent and safe to cache and to serve to any job context. + let used_job_context = std::sync::atomic::AtomicBool::new(false); + let r = transform_json_value_tracked( &db_with_opt_authed, workspace, value, &job_id, token_for_context, 0, + &used_job_context, ) .await?; - if allow_cache { - cache_resource(&workspace, &path, r.clone()); + if let Some(identity) = cache_identity.as_deref() { + if !used_job_context.load(std::sync::atomic::Ordering::Relaxed) { + cache_resource(&workspace, &path, identity, r.clone()); + } } Ok(Some(r)) } else { @@ -601,14 +618,41 @@ pub async fn get_resource_value_interpolated_internal<'a>( // access could otherwise use to crash the API process. pub const MAX_RESOURCE_INTERPOLATION_DEPTH: u8 = 50; -#[async_recursion] pub async fn transform_json_value( - db_with_opt_authed: &DbWithOptAuthed, + db_with_opt_authed: &DbWithOptAuthed<'_, ApiAuthed>, workspace: &str, v: Value, job_id: &Option, token: Option<&str>, depth: u8, +) -> Result { + // Discard the job-context flag; callers that need it use `transform_json_value_tracked`. + let used_job_context = std::sync::atomic::AtomicBool::new(false); + transform_json_value_tracked( + db_with_opt_authed, + workspace, + v, + job_id, + token, + depth, + &used_job_context, + ) + .await +} + +/// Like [`transform_json_value`], but records into `used_job_context` whether the value +/// contains a `$WM_*` contextual variable (resolved from `job_id`/`token`). A value that did +/// not is job-independent and safe to cache; one that did must not be cached or shared across +/// jobs. +#[async_recursion] +pub async fn transform_json_value_tracked( + db_with_opt_authed: &DbWithOptAuthed<'_, ApiAuthed>, + workspace: &str, + v: Value, + job_id: &Option, + token: Option<&str>, + depth: u8, + used_job_context: &std::sync::atomic::AtomicBool, ) -> Result { if depth >= MAX_RESOURCE_INTERPOLATION_DEPTH { return Err(Error::internal_err(format!( @@ -652,15 +696,35 @@ pub async fn transform_json_value( tx.commit().await?; let v = not_found_if_none(v, "Resource", path)?; if let Some(v) = v { - transform_json_value(db_with_opt_authed, workspace, v, job_id, token, depth + 1) - .await + transform_json_value_tracked( + db_with_opt_authed, + workspace, + v, + job_id, + token, + depth + 1, + used_job_context, + ) + .await } else { Ok(Value::Null) } } - Value::String(y) if y.starts_with("$") && job_id.is_some() => { + // `$WM_*` is the reserved contextual-variable namespace (`$WM_TOKEN`, `$WM_JOB_ID`, + // ...); its resolved value depends on the job, so a value containing one is + // job-dependent and must never be cached — including on a no-job read, where the + // placeholder is left unresolved (caching it would then serve a stale placeholder to a + // later job read). Any other `$...` string (custom workspace envs, `$5.00`, `$HOME`, jq + // paths) is NOT interpolated here — it resolves to itself regardless of context and so + // stays cacheable (handled by the catch-all below). Note: custom workspace envs are + // intentionally not resolved inside resource values (they remain available to scripts). + Value::String(y) if y.starts_with("$WM_") => { + used_job_context.store(true, std::sync::atomic::Ordering::Relaxed); + let Some(job_id) = *job_id else { + // No job context to resolve against; leave the placeholder unchanged. + return Ok(Value::String(y)); + }; let mut tx = db_with_opt_authed.begin().await?; - let job_id = job_id.unwrap(); let job = sqlx::query!( "SELECT v2_job.permissioned_as_email, @@ -731,13 +795,14 @@ pub async fn transform_json_value( Value::Array(mut arr) if depth <= 2 && arr.len() <= 1000 => { for i in 0..arr.len() { let val = std::mem::take(&mut arr[i]); - arr[i] = transform_json_value( + arr[i] = transform_json_value_tracked( db_with_opt_authed, workspace, val, job_id, token, depth + 1, + used_job_context, ) .await?; } @@ -754,13 +819,14 @@ pub async fn transform_json_value( } Value::Object(mut m) => { for (a, b) in m.clone().into_iter() { - let v = transform_json_value( + let v = transform_json_value_tracked( db_with_opt_authed, workspace, b, job_id, token, depth + 1, + used_job_context, ) .await?; m.insert(a.clone(), v); diff --git a/backend/windmill-store/src/var_resource_cache.rs b/backend/windmill-store/src/var_resource_cache.rs index f7ce2aeecf..3e89f8579e 100644 --- a/backend/windmill-store/src/var_resource_cache.rs +++ b/backend/windmill-store/src/var_resource_cache.rs @@ -8,7 +8,9 @@ use quick_cache::sync::Cache; use serde_json::Value; +use sha2::{Digest, Sha256}; use std::time::{SystemTime, UNIX_EPOCH}; +use windmill_common::db::Authable; /// Cache TTL for variables and resources (30seconds) const CACHE_TTL_SECS: u64 = 30; @@ -40,11 +42,23 @@ impl CacheEntry { } } -lazy_static::lazy_static! { - /// Cache for individual variable values: key = "workspace_id:path" - pub static ref VARIABLE_CACHE: Cache> = Cache::new(1000); +/// A cached variable value plus whether it is a secret. `is_secret` is retained so a +/// cache hit can re-run the per-read side effects of a secret read (the +/// `variables.decrypt_secret` audit and running-job secret registration) that the +/// original miss performed — a hit must be observably equivalent to a miss. +#[derive(Clone, Debug)] +pub struct CachedVariable { + pub value: String, + pub is_secret: bool, +} - /// Cache for resource values: key = "workspace_id:path" +lazy_static::lazy_static! { + /// Cache for individual variable values. Key: [`identity_cache_key`] + /// (`identity:workspace_id:path`) — scoped to the caller's authorization context. + pub static ref VARIABLE_CACHE: Cache> = Cache::new(1000); + + /// Cache for interpolated resource values. Key: [`identity_cache_key`] + /// (`identity:workspace_id:path`) — scoped to the caller's authorization context. pub static ref RESOURCE_CACHE: Cache> = Cache::new(1000); } @@ -53,9 +67,73 @@ pub fn cache_key(workspace_id: &str, path: &str) -> String { format!("{}:{}", workspace_id, path) } -/// Get cached variable if available and not expired -pub fn get_cached_variable(workspace_id: &str, path: &str) -> Option { - let key = cache_key(workspace_id, path); +/// Hash the caller's full authorization context into a stable identity string. +/// +/// Email alone is **not** a sufficient scope: the same email can resolve to different +/// effective permissions (`username`, groups, folders, scopes, admin/operator) through +/// job- or owner-scoped tokens that share an email but carry a narrower `permissioned_as`. +/// Every input that determines what the caller may read is folded in, mirroring +/// `job_read_access_cache_key` in windmill-api, so a lower-privilege context can never +/// reuse a higher-privilege context's cache entry. Variable-length fields are +/// length-prefixed to keep the encoding injective. +pub fn auth_identity(authed: &A) -> String { + let mut hasher = Sha256::new(); + let field = |hasher: &mut Sha256, bytes: &[u8]| { + hasher.update((bytes.len() as u32).to_be_bytes()); + hasher.update(bytes); + }; + hasher.update([authed.is_admin() as u8, authed.is_operator() as u8]); + field(&mut hasher, authed.email().as_bytes()); + field(&mut hasher, authed.username().as_bytes()); + let mut groups: Vec<&str> = authed.groups().iter().map(String::as_str).collect(); + groups.sort_unstable(); + hasher.update((groups.len() as u32).to_be_bytes()); + for g in groups { + field(&mut hasher, g.as_bytes()); + } + let mut folders: Vec<&str> = authed.folders().iter().map(|f| f.0.as_str()).collect(); + folders.sort_unstable(); + hasher.update((folders.len() as u32).to_be_bytes()); + for f in folders { + field(&mut hasher, f.as_bytes()); + } + match authed.scopes() { + // u32::MAX length-prefix marks "no scopes" so it can't collide with an empty list. + None => hasher.update(u32::MAX.to_be_bytes()), + Some(scopes) => { + let mut scopes: Vec<&str> = scopes.iter().map(String::as_str).collect(); + scopes.sort_unstable(); + hasher.update((scopes.len() as u32).to_be_bytes()); + for s in scopes { + field(&mut hasher, s.as_bytes()); + } + } + } + hex::encode(hasher.finalize()) +} + +/// Generate an identity-scoped cache key (`identity:workspace_id:path`). +/// +/// Both the variable and resource caches store *already-decrypted* values that were +/// resolved under the caller's row-level-security context. The cache is consulted before +/// the per-folder RLS query runs, so an unscoped `workspace:path` key would let an entry +/// warmed by one caller (via `allow_cache=true`) be served to a different caller who has +/// no access to the underlying folder, leaking decrypted secrets within the TTL. `identity` +/// is [`auth_identity`] — the hash of the caller's full authorization context — so a hit +/// can only ever be returned to a caller whose authorized read populated it. +fn identity_cache_key(identity: &str, workspace_id: &str, path: &str) -> String { + format!("{}:{}", identity, cache_key(workspace_id, path)) +} + +/// Get cached variable if available and not expired. Scoped to `identity` +/// ([`auth_identity`]); see [`identity_cache_key`]. Returns the value and its `is_secret` +/// flag so the caller can re-run a secret read's side effects on a hit. +pub fn get_cached_variable( + workspace_id: &str, + path: &str, + identity: &str, +) -> Option { + let key = identity_cache_key(identity, workspace_id, path); VARIABLE_CACHE.get(&key).and_then(|entry| { if entry.is_expired() { VARIABLE_CACHE.remove(&key); @@ -67,17 +145,21 @@ pub fn get_cached_variable(workspace_id: &str, path: &str) -> Option { }) } -/// Cache variable data -pub fn cache_variable(workspace_id: &str, path: &str, email: &str, variable: String) { - let key = format!("{}:{}", email, cache_key(workspace_id, path)); +/// Cache variable data, scoped to the caller identity. See [`get_cached_variable`]. +pub fn cache_variable(workspace_id: &str, path: &str, identity: &str, variable: CachedVariable) { + let key = identity_cache_key(identity, workspace_id, path); let entry = CacheEntry::new(variable); VARIABLE_CACHE.insert(key.clone(), entry); tracing::debug!("Cached variable {}", key); } -/// Get cached resource if available and not expired -pub fn get_cached_resource(workspace_id: &str, path: &str) -> Option { - let key = cache_key(workspace_id, path); +/// Get cached resource if available and not expired. +/// +/// Scoped to `identity` ([`auth_identity`]); see [`identity_cache_key`]. The cached value +/// is the *already-interpolated* resource — its `$var:`/`$res:` secrets are resolved and +/// decrypted inline — so it must never cross authorization boundaries. +pub fn get_cached_resource(workspace_id: &str, path: &str, identity: &str) -> Option { + let key = identity_cache_key(identity, workspace_id, path); RESOURCE_CACHE.get(&key).and_then(|entry| { if entry.is_expired() { RESOURCE_CACHE.remove(&key); @@ -89,22 +171,28 @@ pub fn get_cached_resource(workspace_id: &str, path: &str) -> Option { }) } -/// Cache resource data -pub fn cache_resource(workspace_id: &str, path: &str, resource: Value) { - let key = cache_key(workspace_id, path); +/// Cache resource data, scoped to the caller identity. See [`get_cached_resource`]. +pub fn cache_resource(workspace_id: &str, path: &str, identity: &str, resource: Value) { + let key = identity_cache_key(identity, workspace_id, path); let entry = CacheEntry::new(resource); RESOURCE_CACHE.insert(key.clone(), entry); tracing::debug!("Cached resource {}", key); } -/// Invalidate specific variable from cache +/// Invalidate a variable from the cache. +/// +/// NOTE: entries are keyed by [`identity_cache_key`] (`identity:workspace:path`), so this +/// `workspace:path` key cannot target them — it only removes a legacy unscoped entry, if +/// any. Per-identity entries are not enumerable here; rely on the 30s TTL for staleness, +/// or use [`clear_all_caches`] to force a full flush. Currently unused. pub fn invalidate_variable_cache(workspace_id: &str, path: &str) { let key = cache_key(workspace_id, path); VARIABLE_CACHE.remove(&key); tracing::info!("Variable cache invalidated for {}", key); } -/// Invalidate specific resource from cache +/// Invalidate a resource from the cache. Same identity-scoping caveat as +/// [`invalidate_variable_cache`]. Currently unused. pub fn invalidate_resource_cache(workspace_id: &str, path: &str) { let key = cache_key(workspace_id, path); RESOURCE_CACHE.remove(&key); @@ -118,3 +206,106 @@ pub fn clear_all_caches() { RESOURCE_CACHE.clear(); tracing::debug!("All variable/resource caches cleared"); } + +#[cfg(test)] +mod tests { + use super::*; + + /// Minimal [`Authable`] double so we can assert which authorization fields the + /// cache identity is sensitive to, without standing up a full auth stack. + struct FakeAuthed { + email: String, + username: String, + is_admin: bool, + is_operator: bool, + groups: Vec, + folders: Vec<(String, bool, bool)>, + scopes: Option>, + } + + impl FakeAuthed { + fn base() -> Self { + Self { + email: "alice@x.dev".to_string(), + username: "alice".to_string(), + is_admin: false, + is_operator: false, + groups: vec!["all".to_string()], + folders: vec![("shared".to_string(), false, false)], + scopes: None, + } + } + } + + impl Authable for FakeAuthed { + fn email(&self) -> &str { + &self.email + } + fn username(&self) -> &str { + &self.username + } + fn is_admin(&self) -> bool { + self.is_admin + } + fn is_operator(&self) -> bool { + self.is_operator + } + fn groups(&self) -> &[String] { + &self.groups + } + fn folders(&self) -> &[(String, bool, bool)] { + &self.folders + } + fn scopes(&self) -> Option<&[String]> { + self.scopes.as_deref() + } + } + + // Email alone must NOT determine the cache identity: two contexts that share an email + // but resolve to different effective permissions must get distinct identities, so a + // lower-privilege context can never reuse a higher-privilege one's cached secret. + #[test] + fn auth_identity_is_not_just_email() { + let base = auth_identity(&FakeAuthed::base()); + + let mut more_folders = FakeAuthed::base(); + more_folders + .folders + .push(("secret".to_string(), false, false)); + assert_ne!(base, auth_identity(&more_folders), "folders must matter"); + + let mut more_groups = FakeAuthed::base(); + more_groups.groups.push(("devs").to_string()); + assert_ne!(base, auth_identity(&more_groups), "groups must matter"); + + let mut other_user = FakeAuthed::base(); + other_user.username = "bob".to_string(); + assert_ne!(base, auth_identity(&other_user), "username must matter"); + + let mut admin = FakeAuthed::base(); + admin.is_admin = true; + assert_ne!(base, auth_identity(&admin), "is_admin must matter"); + + let mut operator = FakeAuthed::base(); + operator.is_operator = true; + assert_ne!(base, auth_identity(&operator), "is_operator must matter"); + + let mut scoped = FakeAuthed::base(); + scoped.scopes = Some(vec!["resources:read:f/secret/x".to_string()]); + assert_ne!(base, auth_identity(&scoped), "scopes must matter"); + } + + // Identical authorization contexts must produce the same identity (so the same caller + // gets a cache hit), and ordering of groups/folders must not change the identity. + #[test] + fn auth_identity_is_stable_and_order_independent() { + let a = FakeAuthed::base(); + assert_eq!(auth_identity(&a), auth_identity(&FakeAuthed::base())); + + let mut reordered = FakeAuthed::base(); + reordered.groups = vec!["all".to_string(), "devs".to_string()]; + let mut other_order = FakeAuthed::base(); + other_order.groups = vec!["devs".to_string(), "all".to_string()]; + assert_eq!(auth_identity(&reordered), auth_identity(&other_order)); + } +} diff --git a/backend/windmill-store/src/variables.rs b/backend/windmill-store/src/variables.rs index 2d767b393b..24739ce940 100644 --- a/backend/windmill-store/src/variables.rs +++ b/backend/windmill-store/src/variables.rs @@ -42,7 +42,9 @@ use windmill_common::{ worker::CLOUD_HOSTED, }; -use crate::var_resource_cache::{cache_variable, get_cached_variable}; +use crate::var_resource_cache::{ + auth_identity, cache_variable, get_cached_variable, CachedVariable, +}; use lazy_static::lazy_static; use serde::Deserialize; use sqlx::{Acquire, Postgres, Transaction}; @@ -1204,15 +1206,55 @@ fn replace_path(v: serde_json::Value, path: &str, npath: &str) -> Value { } } +/// Emit the `variables.decrypt_secret` audit event for a secret-variable read. Run on both +/// the cache-miss and cache-hit paths so `allow_cache` never skips secret-access auditing. +async fn audit_decrypt_secret( + db_with_opt_authed: &DbWithOptAuthed<'_, ApiAuthed>, + w_id: &str, + path: &str, +) -> Result<()> { + let mut tx = db_with_opt_authed.db().begin().await?; + audit_log( + &mut *tx, + db_with_opt_authed, + "variables.decrypt_secret", + ActionKind::Execute, + w_id, + Some(path), + None, + ) + .await?; + tx.commit().await?; + Ok(()) +} + pub async fn get_value_internal<'a>( db_with_opt_authed: &'a DbWithOptAuthed<'a, ApiAuthed>, w_id: &str, path: &str, allow_cache: bool, ) -> Result { - if allow_cache { - if let Some(cached_variable) = get_cached_variable(&w_id, &path) { - return Ok(cached_variable); + // Scope the cache to the caller's full authorization identity (not just email): the + // cached value is the decrypted variable, resolved under this caller's RLS context. + let cache_identity = allow_cache.then(|| match db_with_opt_authed.authed() { + Some(authed) => auth_identity(authed), + None => format!("\0system:{}", db_with_opt_authed.email()), + }); + + if let Some(identity) = cache_identity.as_deref() { + if let Some(cached) = get_cached_variable(&w_id, &path, identity) { + // A cache hit must be observably equivalent to a miss: re-run the per-read side + // effects a secret read performs (the `variables.decrypt_secret` audit and + // running-job secret registration) so `allow_cache` never silently skips them. + if cached.is_secret { + audit_decrypt_secret(db_with_opt_authed, &w_id, &path).await?; + if !cached.value.is_empty() { + windmill_common::sensitive_log_masks::register_secret_for_all_running_jobs( + &cached.value, + ); + } + } + return Ok(cached.value); } } @@ -1234,19 +1276,7 @@ pub async fn get_value_internal<'a>( }; let r = if variable.is_secret { - // let audit_author = - let mut tx = db_with_opt_authed.db().begin().await?; - audit_log( - &mut *tx, - db_with_opt_authed, - "variables.decrypt_secret", - ActionKind::Execute, - &w_id, - Some(&variable.path), - None, - ) - .await?; - tx.commit().await?; + audit_decrypt_secret(db_with_opt_authed, &w_id, &variable.path).await?; let value = variable.value; if variable.is_expired.unwrap_or(false) && variable.account.is_some() { @@ -1282,9 +1312,16 @@ pub async fn get_value_internal<'a>( windmill_common::sensitive_log_masks::register_secret_for_all_running_jobs(&r); } - // Cache the result when explicitly allowed and caching appropriate - if allow_cache { - cache_variable(&w_id, &path, db_with_opt_authed.email(), r.clone()); + // Cache the result when explicitly allowed. Secrets are cached too: their per-read side + // effects (audit + running-job registration) are re-run on a hit (see the hit path above), + // and `is_secret` is stored so the hit knows to do so. + if let Some(identity) = cache_identity.as_deref() { + cache_variable( + &w_id, + &path, + identity, + CachedVariable { value: r.clone(), is_secret: variable.is_secret }, + ); } Ok(r)