mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-26 00:01:37 +00:00
Ruben Fiszel
0ba128afe7
The variable and resource value caches (backing
`GET /api/w/{w}/variables/get_value/{path}?allow_cache=true` and
`.../resources/get_value_interpolated/{path}?allow_cache=true`) are consulted
before the per-folder RLS query and store the already-decrypted value. The
resource cache was keyed only by `workspace:path` with no caller identity, so a
cache entry warmed by a privileged peer using `allow_cache=true` could be
returned to a caller with no access to the resource's folder on a cache hit
within the 30s TTL — leaking another folder's decrypted secrets.
Scope both caches to the caller's full authorization identity. The key is now
`auth_identity(authed):workspace:path`, where `auth_identity` is a SHA-256 of the
caller's effective authorization context (email, username, is_admin, is_operator,
sorted groups, sorted folders, sorted scopes) — mirroring
`job_read_access_cache_key`. Email alone is insufficient: the same email can
resolve to different effective permissions via job/owner-scoped tokens, so a
lower-privilege context must not reuse a higher-privilege context's entry.
Job-context resource interpolation is handled correctly: only `$WM_*` contextual
variables are resolved (and only when a `job_id` is present). The interpolation
reports whether the value contains a `$WM_*` placeholder
(`transform_json_value_tracked` + an `AtomicBool`). A value containing one is
job-dependent — even on a no-job read where it's left unresolved — and is never
cached (so a later job read never gets a stale placeholder or another job's
context). Any value without a `$WM_*` placeholder is job-independent and cached
under the identity key, shared across job contexts, so reads carrying a `job_id`
still hit the cache.
BEHAVIOR CHANGE: custom workspace environment variables are no longer interpolated
into resource values via `$NAME` (this was undocumented and prevented caching of
any `$`-prefixed value). Custom envs remain available to scripts/workers as before.
Built-in `$WM_*` contextual variables in resource values are unchanged.
The variable cache previously wrote with an identity-scoped key but read with the
unscoped key, so it never hit (a latent functional bug that happened to be safe).
Aligning the read path enables the cache and makes it identity-scoped by
construction. Secret variables are cached too, but the entry carries the
`is_secret` flag so a cache hit re-runs the per-read side effects a secret read
performs — the EE `variables.decrypt_secret` audit and running-job secret
registration (factored into `audit_decrypt_secret`, shared by both paths).
The unused `invalidate_{variable,resource}_cache` helpers can no longer target
identity-scoped entries; documented the constraint and refreshed the stale
key-format docs on the cache statics.
Tests:
- integration regression for both caches: a folder-scoped user warms the cache via
allow_cache=true, then a user without folder access is denied (401) and never
receives the cached value.
- integration regression that variables (secret included) are served from cache.
- integration regression for job context: plain and non-`$WM_` `$`-string resources
stay cached and are served under a job_id, while a `$WM_*` resource (warmed without
a job_id) is not cached.
- unit tests for `auth_identity`.
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1329 lines
42 KiB
Rust
1329 lines
42 KiB
Rust
/*
|
|
* Author: Ruben Fiszel
|
|
* Copyright: Windmill Labs, Inc 2022
|
|
* This file and its contents are licensed under the AGPLv3 License.
|
|
* Please see the included NOTICE for copyright information and
|
|
* LICENSE-AGPL for a copy of the license.
|
|
*/
|
|
|
|
use windmill_api_auth::{
|
|
build_scope_path_predicate, check_scopes, maybe_refresh_folders, require_owner_of_path,
|
|
ApiAuthed,
|
|
};
|
|
use windmill_common::db::DB;
|
|
use windmill_common::workspaces::{check_deploy_rules, RuleCheckResult};
|
|
|
|
use crate::secret_backend_ext::{
|
|
delete_secret_from_backend, get_secret_value, is_vault_stored_value, rename_vault_secret,
|
|
store_secret_value,
|
|
};
|
|
use windmill_common::utils::{escape_ilike_pattern, BulkDeleteRequest};
|
|
use windmill_common::webhook::{WebhookMessage, WebhookShared};
|
|
|
|
use axum::{
|
|
extract::{Extension, Path, Query},
|
|
routing::{delete, get, post},
|
|
Json, Router,
|
|
};
|
|
use futures::future::try_join_all;
|
|
use hyper::StatusCode;
|
|
use serde_json::Value;
|
|
|
|
use windmill_audit::audit_oss::{audit_log, AuditAuthorable};
|
|
use windmill_audit::ActionKind;
|
|
use windmill_common::{
|
|
db::{DbWithOptAuthed, UserDB},
|
|
error::{Error, JsonResult, Result},
|
|
scripts::ScriptHash,
|
|
utils::{not_found_if_none, paginate, Pagination, StripPath, WarnAfterExt},
|
|
variables::{
|
|
build_crypt, get_reserved_variables, ContextualVariable, CreateVariable, ListableVariable,
|
|
},
|
|
worker::CLOUD_HOSTED,
|
|
};
|
|
|
|
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};
|
|
use windmill_common::variables::encrypt;
|
|
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
|
|
|
|
lazy_static! {
|
|
pub static ref SECRET_SALT: Option<String> = std::env::var("SECRET_SALT").ok();
|
|
}
|
|
|
|
pub fn workspaced_service() -> Router {
|
|
Router::new()
|
|
.route("/list", get(list_variables))
|
|
.route("/list_contextual", get(list_contextual_variables))
|
|
.route("/get/{*path}", get(get_variable))
|
|
.route("/get_value/{*path}", get(get_value))
|
|
.route("/exists/{*path}", get(exists_variable))
|
|
.route("/update/{*path}", post(update_variable))
|
|
.route("/delete/{*path}", delete(delete_variable))
|
|
.route("/delete_bulk", delete(delete_variables_bulk))
|
|
.route("/create", post(create_variable))
|
|
.route("/encrypt", post(encrypt_value))
|
|
}
|
|
|
|
async fn list_contextual_variables(
|
|
Path(w_id): Path<String>,
|
|
ApiAuthed { username, email, .. }: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
) -> JsonResult<Vec<ContextualVariable>> {
|
|
Ok(Json(
|
|
get_reserved_variables(
|
|
&db.into(),
|
|
&w_id,
|
|
"q1A0qcPuO00yxioll7iph76N9CJDqn",
|
|
&email,
|
|
&username,
|
|
"017e0ad5-f499-73b6-5488-92a61c5196dd",
|
|
format!("u/{username}").as_str(),
|
|
Some("u/user/script_path".to_string()),
|
|
Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()),
|
|
Some("u/user/encapsulating_flow_path".to_string()),
|
|
Some("u/user/triggering_flow_path".to_string()),
|
|
Some("c".to_string()),
|
|
Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()),
|
|
Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()),
|
|
Some(chrono::offset::Utc::now()),
|
|
Some(ScriptHash(1234567890)),
|
|
None,
|
|
None,
|
|
)
|
|
.await
|
|
.to_vec(),
|
|
))
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct ListVariableQuery {
|
|
pub path_start: Option<String>,
|
|
pub path: Option<String>,
|
|
pub description: Option<String>,
|
|
// filter by matching the non-encrypted value (for non-secrets only)
|
|
pub value: Option<String>,
|
|
pub broad_filter: Option<String>,
|
|
pub label: Option<String>,
|
|
}
|
|
|
|
async fn list_variables(
|
|
authed: ApiAuthed,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Path(w_id): Path<String>,
|
|
Query(lq): Query<ListVariableQuery>,
|
|
Query(pagination): Query<Pagination>,
|
|
) -> JsonResult<Vec<ListableVariable>> {
|
|
use sql_builder::{bind::Bind, SqlBuilder};
|
|
|
|
let (per_page, offset) = paginate(pagination);
|
|
|
|
let mut sqlb = SqlBuilder::select_from("variable")
|
|
.fields(&[
|
|
"variable.workspace_id",
|
|
"variable.path",
|
|
"CASE WHEN is_secret IS TRUE THEN null ELSE variable.value::text END as value",
|
|
"is_secret",
|
|
"variable.description",
|
|
"variable.extra_perms",
|
|
"account",
|
|
"is_oauth",
|
|
"(now() > account.expires_at) as is_expired",
|
|
"account.refresh_error",
|
|
"resource.path IS NOT NULL as is_linked",
|
|
"account.refresh_token != '' as is_refreshed",
|
|
"variable.expires_at",
|
|
"variable.labels",
|
|
"ws_specific.path IS NOT NULL as ws_specific",
|
|
"variable.edited_at",
|
|
"variable.edited_by",
|
|
])
|
|
.left()
|
|
.join("account")
|
|
.on("variable.account = account.id AND account.workspace_id = ?".bind(&w_id))
|
|
.left()
|
|
.join("resource")
|
|
.on("resource.path = variable.path AND resource.workspace_id = ?".bind(&w_id))
|
|
.left()
|
|
.join("ws_specific")
|
|
.on("ws_specific.path = variable.path AND ws_specific.workspace_id = variable.workspace_id AND ws_specific.item_kind = 'variable'")
|
|
.and_where("variable.workspace_id = ?".bind(&w_id))
|
|
.and_where("variable.path NOT LIKE 'u/' || ? || '/secret_arg/%'".bind(&authed.username))
|
|
.order_by("path", false)
|
|
.limit(per_page)
|
|
.offset(offset)
|
|
.clone();
|
|
|
|
if let Some(path_start) = &lq.path_start {
|
|
sqlb.and_where_like_left("variable.path", path_start);
|
|
}
|
|
|
|
if let Some(path) = &lq.path {
|
|
sqlb.and_where_eq("variable.path", "?".bind(path));
|
|
}
|
|
|
|
if let Some(description) = &lq.description {
|
|
let pat = format!("%{}%", escape_ilike_pattern(description));
|
|
sqlb.and_where("variable.description ILIKE ?".bind(&pat));
|
|
}
|
|
|
|
if let Some(value) = &lq.value {
|
|
// Only filter on non-secret variables' value
|
|
let pat = format!("%{}%", escape_ilike_pattern(value));
|
|
sqlb.and_where("(is_secret = FALSE AND variable.value ILIKE ?)".bind(&pat));
|
|
}
|
|
|
|
if let Some(broad_filter) = &lq.broad_filter {
|
|
let pat = format!("%{}%", escape_ilike_pattern(broad_filter));
|
|
sqlb.and_where(
|
|
"(variable.path ILIKE ? OR variable.description ILIKE ? OR (is_secret = FALSE AND variable.value ILIKE ?))"
|
|
.bind(&pat).bind(&pat).bind(&pat)
|
|
);
|
|
}
|
|
|
|
if let Some(label) = &lq.label {
|
|
for l in label.split(',') {
|
|
sqlb.and_where("variable.labels @> ARRAY[?]".bind(&l.trim()));
|
|
}
|
|
}
|
|
|
|
let sql = sqlb.sql().map_err(|e| Error::internal_err(e.to_string()))?;
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
let allowed = build_scope_path_predicate(&authed, "variables", "read");
|
|
let rows = sqlx::query_as::<_, ListableVariable>(&sql)
|
|
.fetch_all(&mut *tx)
|
|
.await?
|
|
.into_iter()
|
|
.filter(|r| allowed(&r.path))
|
|
.collect::<Vec<_>>();
|
|
|
|
tx.commit().await?;
|
|
Ok(Json(rows))
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct GetVariableQuery {
|
|
decrypt_secret: Option<bool>,
|
|
include_encrypted: Option<bool>,
|
|
}
|
|
|
|
async fn get_variable(
|
|
authed: ApiAuthed,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Extension(db): Extension<DB>,
|
|
Query(q): Query<GetVariableQuery>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
) -> JsonResult<ListableVariable> {
|
|
let path = path.to_path();
|
|
check_scopes(&authed, || format!("variables:read:{}", path))?;
|
|
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
|
|
let variable_o = sqlx::query_as::<_, ListableVariable>(
|
|
"SELECT variable.workspace_id, variable.path, variable.value, variable.is_secret,
|
|
variable.description, variable.extra_perms, variable.account, variable.is_oauth,
|
|
variable.expires_at, variable.labels,
|
|
variable.edited_at, variable.edited_by,
|
|
(now() > account.expires_at) as is_expired, account.refresh_error,
|
|
resource.path IS NOT NULL as is_linked,
|
|
account.refresh_token != '' as is_refreshed,
|
|
ws_specific.path IS NOT NULL as ws_specific
|
|
FROM variable
|
|
LEFT JOIN account ON variable.account = account.id
|
|
LEFT JOIN resource ON resource.path = variable.path AND resource.workspace_id = $2
|
|
LEFT JOIN ws_specific ON ws_specific.path = variable.path AND ws_specific.workspace_id = $2 AND ws_specific.item_kind = 'variable'
|
|
WHERE variable.path = $1 AND variable.workspace_id = $2
|
|
LIMIT 1",
|
|
)
|
|
.bind(&path)
|
|
.bind(&w_id)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
let variable = if let Some(variable) = variable_o {
|
|
variable
|
|
} else {
|
|
explain_variable_perm_error(&path, &w_id, &db).await?;
|
|
unreachable!()
|
|
};
|
|
|
|
let decrypt_secret = q.decrypt_secret.unwrap_or(true);
|
|
|
|
let r = if variable.is_secret {
|
|
if decrypt_secret {
|
|
audit_log(
|
|
&mut *tx,
|
|
&authed,
|
|
"variables.decrypt_secret",
|
|
ActionKind::Execute,
|
|
&w_id,
|
|
Some(&variable.path),
|
|
None,
|
|
)
|
|
.await?;
|
|
}
|
|
|
|
let value = variable.value.unwrap_or_else(|| "".to_string());
|
|
ListableVariable {
|
|
value: if variable.is_expired.unwrap_or(false) && variable.account.is_some() {
|
|
#[cfg(feature = "oauth2")]
|
|
{
|
|
let refresh_tx = db
|
|
.begin()
|
|
.await
|
|
.map_err(|e| Error::InternalErr(e.to_string()))?;
|
|
Some(
|
|
crate::oauth_refresh_oss::_refresh_token(
|
|
refresh_tx,
|
|
&variable.path,
|
|
&w_id,
|
|
variable.account.unwrap(),
|
|
&db,
|
|
)
|
|
.await?,
|
|
)
|
|
}
|
|
#[cfg(not(feature = "oauth2"))]
|
|
return Err(Error::internal_err("Require oauth2 feature".to_string()));
|
|
} else if !value.is_empty() && decrypt_secret {
|
|
let _ = tx.commit().await;
|
|
// Use secret backend for decryption (supports both DB and Vault)
|
|
Some(get_secret_value(&db, &w_id, &variable.path, &value).await?)
|
|
} else if q.include_encrypted.unwrap_or(false) {
|
|
Some(value)
|
|
} else {
|
|
None
|
|
},
|
|
..variable
|
|
}
|
|
} else {
|
|
variable
|
|
};
|
|
|
|
Ok(Json(r))
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct GetValueQuery {
|
|
allow_cache: Option<bool>,
|
|
}
|
|
async fn get_value(
|
|
authed: ApiAuthed,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Extension(db): Extension<DB>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
Query(q): Query<GetValueQuery>,
|
|
) -> JsonResult<String> {
|
|
let path = path.to_path();
|
|
check_scopes(&authed, || format!("variables:read:{}", path))?;
|
|
let userdb_authed = DbWithOptAuthed::from_authed(&authed, db.clone(), Some(user_db.clone()));
|
|
|
|
return get_value_internal(&userdb_authed, &w_id, &path, q.allow_cache.unwrap_or(false))
|
|
.warn_after_seconds(10)
|
|
.await
|
|
.map(Json);
|
|
}
|
|
|
|
async fn explain_variable_perm_error(
|
|
path: &str,
|
|
w_id: &str,
|
|
db: &sqlx::Pool<Postgres>,
|
|
) -> windmill_common::error::Result<()> {
|
|
let extra_perms = sqlx::query_scalar!(
|
|
"SELECT extra_perms from variable WHERE path = $1 AND workspace_id = $2",
|
|
path,
|
|
w_id
|
|
)
|
|
.fetch_optional(db)
|
|
.await?
|
|
.ok_or_else(|| Error::NotFound(format!("Variable {} not found", path)))?;
|
|
if path.starts_with("f/") {
|
|
let folder = path.split("/").nth(1).ok_or_else(|| {
|
|
Error::BadRequest(format!(
|
|
"path {} should have at least 2 components separated by /",
|
|
path
|
|
))
|
|
})?;
|
|
let folder_extra_perms = sqlx::query_scalar!(
|
|
"SELECT extra_perms from folder WHERE name = $1 AND workspace_id = $2",
|
|
folder,
|
|
w_id
|
|
)
|
|
.fetch_optional(db)
|
|
.await?;
|
|
return Err(Error::NotAuthorized(format!(
|
|
"Variable exists but you don't have access to it:\nvariable perms: {}\nfolder perms: {}",
|
|
serde_json::to_string_pretty(&extra_perms).unwrap_or_default(), serde_json::to_string_pretty(&folder_extra_perms).unwrap_or_default()
|
|
)));
|
|
} else {
|
|
return Err(Error::NotAuthorized(format!(
|
|
"Variable exists but you don't have access to it:\nvariable perms: {}",
|
|
serde_json::to_string_pretty(&extra_perms).unwrap_or_default()
|
|
)));
|
|
}
|
|
}
|
|
|
|
async fn exists_variable(
|
|
Extension(db): Extension<DB>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
) -> JsonResult<bool> {
|
|
let path = path.to_path();
|
|
|
|
let exists = sqlx::query_scalar!(
|
|
"SELECT EXISTS(SELECT 1 FROM variable WHERE path = $1 AND workspace_id = $2)",
|
|
path,
|
|
w_id
|
|
)
|
|
.fetch_one(&db)
|
|
.await?
|
|
.unwrap_or(false);
|
|
|
|
Ok(Json(exists))
|
|
}
|
|
|
|
async fn check_path_conflict(db: &DB, w_id: &str, path: &str) -> Result<()> {
|
|
let exists = sqlx::query_scalar!(
|
|
"SELECT EXISTS(SELECT 1 FROM variable WHERE path = $1 AND workspace_id = $2)",
|
|
path,
|
|
w_id
|
|
)
|
|
.fetch_one(db)
|
|
.await?
|
|
.unwrap_or(false);
|
|
if exists {
|
|
return Err(Error::BadRequest(format!(
|
|
"Variable {} already exists",
|
|
path
|
|
)));
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
async fn create_variable(
|
|
authed: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Extension(webhook): Extension<WebhookShared>,
|
|
Path(w_id): Path<String>,
|
|
Query(AlreadyEncrypted { already_encrypted }): Query<AlreadyEncrypted>,
|
|
Json(variable): Json<CreateVariable>,
|
|
) -> Result<(StatusCode, String)> {
|
|
check_scopes(&authed, || format!("variables:write:{}", variable.path))?;
|
|
if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
|
|
&w_id,
|
|
AuditAuthorable::username(&authed),
|
|
&authed.groups,
|
|
authed.is_admin,
|
|
&db,
|
|
)
|
|
.await?
|
|
{
|
|
return Err(Error::PermissionDenied(msg));
|
|
}
|
|
|
|
if *CLOUD_HOSTED {
|
|
let nb_variables = sqlx::query_scalar!(
|
|
"SELECT COUNT(*) FROM variable WHERE workspace_id = $1",
|
|
&w_id
|
|
)
|
|
.fetch_one(&db)
|
|
.await?;
|
|
if nb_variables.unwrap_or(0) >= 10000 {
|
|
return Err(Error::BadRequest(
|
|
"You have reached the maximum number of variables (10000) on cloud. Check your usage in Workspace Settings > General > Cloud Quotas. Contact support@windmill.dev to increase the limit"
|
|
.to_string(),
|
|
));
|
|
}
|
|
}
|
|
let authed = maybe_refresh_folders(&variable.path, &w_id, authed, &db).await;
|
|
|
|
check_path_conflict(&db, &w_id, &variable.path).await?;
|
|
let value = if variable.is_secret && !already_encrypted.unwrap_or(false) {
|
|
// Use secret backend for encryption (supports both DB and Vault)
|
|
store_secret_value(&db, &w_id, &variable.path, &variable.value).await?
|
|
} else {
|
|
variable.value
|
|
};
|
|
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
|
|
sqlx::query!(
|
|
"INSERT INTO variable
|
|
(workspace_id, path, value, is_secret, description, account, is_oauth, expires_at, labels, edited_by)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
|
|
&w_id,
|
|
variable.path,
|
|
value,
|
|
variable.is_secret,
|
|
variable.description,
|
|
variable.account,
|
|
variable.is_oauth.unwrap_or(false),
|
|
variable.expires_at,
|
|
variable.labels.as_deref() as Option<&[String]>,
|
|
&authed.username
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
if variable.ws_specific.unwrap_or(false) {
|
|
sqlx::query!(
|
|
"INSERT INTO ws_specific (workspace_id, item_kind, path) VALUES ($1, 'variable', $2) ON CONFLICT DO NOTHING",
|
|
w_id,
|
|
variable.path,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
}
|
|
|
|
audit_log(
|
|
&mut *tx,
|
|
&authed,
|
|
"variables.create",
|
|
ActionKind::Create,
|
|
&w_id,
|
|
Some(&variable.path),
|
|
None,
|
|
)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Variable { path: variable.path.clone(), parent_path: None },
|
|
Some(format!("Variable '{}' created", variable.path.clone())),
|
|
true,
|
|
None,
|
|
)
|
|
.await?;
|
|
|
|
webhook.send_message(
|
|
w_id.clone(),
|
|
WebhookMessage::CreateVariable { workspace: w_id, path: variable.path.clone() },
|
|
);
|
|
|
|
Ok((
|
|
StatusCode::CREATED,
|
|
format!("variable {} created", variable.path),
|
|
))
|
|
}
|
|
|
|
async fn encrypt_value(
|
|
Extension(db): Extension<DB>,
|
|
Path(w_id): Path<String>,
|
|
Json(variable): Json<String>,
|
|
) -> Result<String> {
|
|
let mc = build_crypt(&db, &w_id).await?;
|
|
let value = encrypt(&mc, &variable);
|
|
|
|
Ok(value)
|
|
}
|
|
|
|
async fn delete_variable(
|
|
authed: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Extension(webhook): Extension<WebhookShared>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
) -> Result<String> {
|
|
let path = path.to_path();
|
|
|
|
check_scopes(&authed, || format!("variables:write:{}", path))?;
|
|
if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
|
|
&w_id,
|
|
AuditAuthorable::username(&authed),
|
|
&authed.groups,
|
|
authed.is_admin,
|
|
&db,
|
|
)
|
|
.await?
|
|
{
|
|
return Err(Error::PermissionDenied(msg));
|
|
}
|
|
|
|
// Check if variable is a secret before deleting (for Vault cleanup)
|
|
let is_secret = sqlx::query_scalar!(
|
|
"SELECT is_secret FROM variable WHERE path = $1 AND workspace_id = $2",
|
|
path,
|
|
&w_id
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.unwrap_or(false);
|
|
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
|
|
// Capture data for trashbin before deleting
|
|
let trash_var: Option<serde_json::Value> = sqlx::query_scalar(
|
|
"SELECT to_jsonb(t) FROM variable t WHERE path = $1 AND workspace_id = $2",
|
|
)
|
|
.bind(path)
|
|
.bind(&w_id)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
let trash_linked_resource: Option<serde_json::Value> = sqlx::query_scalar(
|
|
"SELECT to_jsonb(t) FROM resource t WHERE path = $1 AND workspace_id = $2",
|
|
)
|
|
.bind(path)
|
|
.bind(&w_id)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
sqlx::query!(
|
|
"DELETE FROM variable WHERE path = $1 AND workspace_id = $2",
|
|
path,
|
|
w_id
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
// Clean up any ws_specific row for the linked resource at the same path
|
|
// before deleting the resource itself — symmetric with the cleanup
|
|
// delete_resource does for linked variables. Without this, a resource
|
|
// later recreated at the same path would inherit a stale ws_specific flag.
|
|
sqlx::query!(
|
|
"DELETE FROM ws_specific
|
|
WHERE workspace_id = $1 AND item_kind = 'resource' AND path = $2",
|
|
w_id,
|
|
path
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
let deleted_linked_resource = sqlx::query_scalar!(
|
|
"DELETE FROM resource WHERE path = $1 AND workspace_id = $2 RETURNING path",
|
|
path,
|
|
w_id
|
|
)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
if let Some(var_data) = trash_var {
|
|
let mut trash_data = serde_json::json!({"row": var_data});
|
|
if let Some(linked) = trash_linked_resource {
|
|
trash_data["linked_resource"] = linked;
|
|
}
|
|
windmill_common::trashbin::move_to_trash(
|
|
&mut *tx,
|
|
&w_id,
|
|
"variable",
|
|
path,
|
|
trash_data,
|
|
&authed.username,
|
|
)
|
|
.await?;
|
|
}
|
|
|
|
sqlx::query!(
|
|
"DELETE FROM ws_specific WHERE workspace_id = $1 AND item_kind = 'variable' AND path = $2",
|
|
w_id,
|
|
path
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
audit_log(
|
|
&mut *tx,
|
|
&authed,
|
|
"variables.delete",
|
|
ActionKind::Delete,
|
|
&w_id,
|
|
Some(path),
|
|
None,
|
|
)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
// If variable was a secret, also delete from Vault backend (if configured)
|
|
if is_secret {
|
|
delete_secret_from_backend(&db, &w_id, path).await?;
|
|
}
|
|
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Variable { path: path.to_string(), parent_path: Some(path.to_string()) },
|
|
Some(format!("Variable '{}' deleted", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?;
|
|
|
|
webhook.send_message(
|
|
w_id.clone(),
|
|
WebhookMessage::DeleteVariable { workspace: w_id.clone(), path: path.to_owned() },
|
|
);
|
|
|
|
if deleted_linked_resource.is_some() {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Resource {
|
|
path: path.to_string(),
|
|
parent_path: Some(path.to_string()),
|
|
},
|
|
Some(format!(
|
|
"Resource '{}' deleted (linked variable deleted)",
|
|
path
|
|
)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?;
|
|
|
|
webhook.send_message(
|
|
w_id.clone(),
|
|
WebhookMessage::DeleteResource { workspace: w_id, path: path.to_owned() },
|
|
);
|
|
}
|
|
|
|
Ok(format!("variable {} deleted", path))
|
|
}
|
|
|
|
async fn delete_variables_bulk(
|
|
authed: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Extension(webhook): Extension<WebhookShared>,
|
|
Path(w_id): Path<String>,
|
|
Json(request): Json<BulkDeleteRequest>,
|
|
) -> JsonResult<Vec<String>> {
|
|
for path in &request.paths {
|
|
check_scopes(&authed, || format!("variables:write:{}", path))?;
|
|
}
|
|
|
|
if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
|
|
&w_id,
|
|
AuditAuthorable::username(&authed),
|
|
&authed.groups,
|
|
authed.is_admin,
|
|
&db,
|
|
)
|
|
.await?
|
|
{
|
|
return Err(Error::PermissionDenied(msg));
|
|
}
|
|
|
|
// Query which paths are secrets before deletion (for Vault cleanup)
|
|
let secret_paths: Vec<String> = sqlx::query_scalar!(
|
|
"SELECT path FROM variable WHERE path = ANY($1) AND workspace_id = $2 AND is_secret = true",
|
|
&request.paths,
|
|
&w_id
|
|
)
|
|
.fetch_all(&db)
|
|
.await?;
|
|
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
|
|
// Capture variables for trashbin per path before bulk delete
|
|
for path in &request.paths {
|
|
let trash_var: Option<serde_json::Value> = sqlx::query_scalar(
|
|
"SELECT to_jsonb(t) FROM variable t WHERE path = $1 AND workspace_id = $2",
|
|
)
|
|
.bind(path)
|
|
.bind(&w_id)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
if let Some(var_data) = trash_var {
|
|
let trash_linked: Option<serde_json::Value> = sqlx::query_scalar(
|
|
"SELECT to_jsonb(t) FROM resource t WHERE path = $1 AND workspace_id = $2",
|
|
)
|
|
.bind(path)
|
|
.bind(&w_id)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
let mut trash_data = serde_json::json!({"row": var_data});
|
|
if let Some(linked) = trash_linked {
|
|
trash_data["linked_resource"] = linked;
|
|
}
|
|
windmill_common::trashbin::move_to_trash(
|
|
&mut *tx,
|
|
&w_id,
|
|
"variable",
|
|
path,
|
|
trash_data,
|
|
&authed.username,
|
|
)
|
|
.await?;
|
|
}
|
|
}
|
|
|
|
let deleted_paths = sqlx::query_scalar!(
|
|
"DELETE FROM variable WHERE path = ANY($1) AND workspace_id = $2 RETURNING path",
|
|
&request.paths,
|
|
w_id
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await?;
|
|
// Mirror single delete_variable: clean the linked-resource ws_specific
|
|
// markers BEFORE deleting the resource rows so they don't survive as
|
|
// orphans. A resource later created at the same path would otherwise
|
|
// inherit a stale ws_specific flag.
|
|
sqlx::query!(
|
|
"DELETE FROM ws_specific WHERE workspace_id = $1 AND item_kind = 'resource' AND path = ANY($2)",
|
|
w_id,
|
|
&deleted_paths
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
sqlx::query!(
|
|
"DELETE FROM resource WHERE path = ANY($1) AND workspace_id = $2",
|
|
&deleted_paths,
|
|
w_id
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
sqlx::query!(
|
|
"DELETE FROM ws_specific WHERE workspace_id = $1 AND item_kind = 'variable' AND path = ANY($2)",
|
|
w_id,
|
|
&deleted_paths
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
audit_log(
|
|
&mut *tx,
|
|
&authed,
|
|
"variables.delete_bulk",
|
|
ActionKind::Delete,
|
|
&w_id,
|
|
Some(&deleted_paths.join(", ")),
|
|
None,
|
|
)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
// Delete secrets from Vault backend (if configured)
|
|
for path in &secret_paths {
|
|
if deleted_paths.contains(path) {
|
|
delete_secret_from_backend(&db, &w_id, path).await?;
|
|
}
|
|
}
|
|
|
|
try_join_all(deleted_paths.iter().map(|path| {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Variable {
|
|
path: path.to_string(),
|
|
parent_path: Some(path.to_string()),
|
|
},
|
|
Some(format!("Variable '{}' deleted", path)),
|
|
true,
|
|
None,
|
|
)
|
|
}))
|
|
.await?;
|
|
|
|
for path in &deleted_paths {
|
|
webhook.send_message(
|
|
w_id.clone(),
|
|
WebhookMessage::DeleteVariable { workspace: w_id.clone(), path: path.to_owned() },
|
|
);
|
|
}
|
|
|
|
Ok(Json(deleted_paths))
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct EditVariable {
|
|
path: Option<String>,
|
|
value: Option<String>,
|
|
is_secret: Option<bool>,
|
|
description: Option<String>,
|
|
account: Option<i32>,
|
|
labels: Option<Vec<String>>,
|
|
ws_specific: Option<bool>,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct AlreadyEncrypted {
|
|
already_encrypted: Option<bool>,
|
|
}
|
|
|
|
async fn update_variable(
|
|
authed: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Extension(webhook): Extension<WebhookShared>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
Query(AlreadyEncrypted { already_encrypted }): Query<AlreadyEncrypted>,
|
|
Json(ns): Json<EditVariable>,
|
|
) -> Result<String> {
|
|
use sql_builder::prelude::*;
|
|
|
|
if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
|
|
&w_id,
|
|
AuditAuthorable::username(&authed),
|
|
&authed.groups,
|
|
authed.is_admin,
|
|
&db,
|
|
)
|
|
.await?
|
|
{
|
|
return Err(Error::PermissionDenied(msg));
|
|
}
|
|
|
|
let path = path.to_path();
|
|
check_scopes(&authed, || format!("variables:write:{}", path))?;
|
|
let authed = maybe_refresh_folders(&path, &w_id, authed, &db).await;
|
|
|
|
let mut sqlb = SqlBuilder::update_table("variable");
|
|
let mut has_sql_updates = false;
|
|
sqlb.and_where_eq("path", "?".bind(&path));
|
|
sqlb.and_where_eq("workspace_id", "?".bind(&w_id));
|
|
|
|
if let Some(npath) = &ns.path {
|
|
sqlb.set_str("path", npath);
|
|
has_sql_updates = true;
|
|
}
|
|
let ns_value_is_none = ns.value.is_none();
|
|
// Determine the target path for storing secrets (use new path if provided)
|
|
let target_path = ns.path.as_deref().unwrap_or(path);
|
|
if let Some(nvalue) = ns.value.clone() {
|
|
let is_secret = if ns.is_secret.is_some() {
|
|
ns.is_secret.unwrap()
|
|
} else {
|
|
sqlx::query_scalar!(
|
|
"SELECT is_secret from variable WHERE path = $1 AND workspace_id = $2",
|
|
&path,
|
|
&w_id
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.unwrap_or(false)
|
|
};
|
|
|
|
let value = if is_secret && !already_encrypted.unwrap_or(false) {
|
|
// Use secret backend for encryption (supports both DB and Vault)
|
|
// Store at target_path (new path if renaming, otherwise current path)
|
|
store_secret_value(&db, &w_id, target_path, &nvalue).await?
|
|
} else {
|
|
nvalue
|
|
};
|
|
sqlb.set_str("value", &value);
|
|
has_sql_updates = true;
|
|
}
|
|
|
|
if let Some(desc) = ns.description {
|
|
sqlb.set_str("description", &desc);
|
|
has_sql_updates = true;
|
|
}
|
|
|
|
if let Some(account_id) = ns.account {
|
|
sqlb.set_str("account", account_id);
|
|
has_sql_updates = true;
|
|
}
|
|
|
|
if let Some(nbool) = ns.is_secret {
|
|
let old_secret = sqlx::query_scalar!(
|
|
"SELECT is_secret from variable WHERE path = $1 AND workspace_id = $2",
|
|
&path,
|
|
&w_id
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.unwrap_or(false);
|
|
if old_secret != nbool && ns_value_is_none {
|
|
return Err(Error::BadRequest(
|
|
"cannot change is_secret without updating value too".to_string(),
|
|
));
|
|
}
|
|
sqlb.set_str("is_secret", nbool);
|
|
has_sql_updates = true;
|
|
}
|
|
|
|
// Get old account_id if we're updating the account field
|
|
let old_account_id = if ns.account.is_some() {
|
|
sqlx::query_scalar!(
|
|
"SELECT account FROM variable WHERE path = $1 AND workspace_id = $2",
|
|
&path,
|
|
&w_id
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.flatten()
|
|
} else {
|
|
None
|
|
};
|
|
|
|
let mut tx: Transaction<'_, Postgres> = user_db.begin(&authed).await?;
|
|
|
|
if let Some(npath) = ns.path.clone() {
|
|
if npath != path {
|
|
check_path_conflict(&db, &w_id, &npath).await?;
|
|
require_owner_of_path(&authed, path)?;
|
|
|
|
// Handle Vault secret rename if the variable is a secret stored in Vault
|
|
let current_var = sqlx::query!(
|
|
"SELECT value, is_secret FROM variable WHERE path = $1 AND workspace_id = $2",
|
|
path,
|
|
w_id
|
|
)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
if let Some(var) = current_var {
|
|
if var.is_secret && is_vault_stored_value(&var.value) {
|
|
if ns.value.is_some() {
|
|
// New value was provided and already stored at new path
|
|
// Just delete the old secret from Vault
|
|
delete_secret_from_backend(&db, &w_id, path).await?;
|
|
} else {
|
|
// No new value - rename the secret in Vault
|
|
if let Some(new_value) =
|
|
rename_vault_secret(&db, &w_id, path, &npath, &var.value).await?
|
|
{
|
|
// Update the variable's value to point to the new Vault path
|
|
sqlb.set_str("value", &new_value);
|
|
has_sql_updates = true;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut v = sqlx::query_scalar!(
|
|
"SELECT value FROM resource WHERE path = $1 AND workspace_id = $2",
|
|
path,
|
|
w_id
|
|
)
|
|
.fetch_optional(&mut *tx)
|
|
.await?
|
|
.flatten();
|
|
|
|
if let Some(old_v) = v {
|
|
v = Some(replace_path(
|
|
old_v,
|
|
&format!("$var:{path}"),
|
|
&format!("$var:{npath}"),
|
|
))
|
|
}
|
|
|
|
sqlx::query!(
|
|
"UPDATE resource SET path = $1, value = $2, edited_at = now() WHERE path = $3 AND workspace_id = $4",
|
|
npath,
|
|
v,
|
|
path,
|
|
w_id
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
sqlx::query!(
|
|
"UPDATE workspace_integrations SET resource_path = $1 WHERE workspace_id = $2 AND resource_path = $3",
|
|
npath,
|
|
w_id,
|
|
path
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
sqlx::query!(
|
|
"UPDATE ws_specific SET path = $1 WHERE workspace_id = $2 AND item_kind = 'variable' AND path = $3",
|
|
npath,
|
|
w_id,
|
|
path
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
// The linked resource at the same path is renamed above; move
|
|
// its ws_specific 'resource' marker too so an explicitly-flagged
|
|
// resource doesn't lose its ws_specific status on rename and
|
|
// doesn't leave a stale marker at the old path. Symmetric with
|
|
// update_resource's rename block.
|
|
sqlx::query!(
|
|
"UPDATE ws_specific SET path = $1 WHERE workspace_id = $2 AND item_kind = 'resource' AND path = $3",
|
|
npath,
|
|
w_id,
|
|
path
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
}
|
|
}
|
|
|
|
let npath = if has_sql_updates {
|
|
sqlb.set("edited_at", "now()");
|
|
sqlb.set_str("edited_by", &authed.username);
|
|
sqlb.returning("path");
|
|
let sql = sqlb.sql().map_err(|e| Error::internal_err(e.to_string()))?;
|
|
let npath_o: Option<String> = sqlx::query_scalar(&sql).fetch_optional(&mut *tx).await?;
|
|
not_found_if_none(npath_o, "Variable", path)?
|
|
} else {
|
|
// `has_sql_updates` is guaranteed true whenever `ns.path` is provided
|
|
// (set unconditionally above when ns.path is Some). So this branch
|
|
// only runs for non-rename edits (e.g. labels-only or ws_specific-only)
|
|
// and the rename-side queries (secret table, variable_history,
|
|
// ws_specific path UPDATE) in the rename block above are skipped.
|
|
debug_assert!(
|
|
ns.path.is_none(),
|
|
"has_sql_updates should be true when ns.path is Some"
|
|
);
|
|
let exists = sqlx::query_scalar::<_, bool>(
|
|
"SELECT EXISTS(SELECT 1 FROM variable WHERE path = $1 AND workspace_id = $2)",
|
|
)
|
|
.bind(path)
|
|
.bind(&w_id)
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
not_found_if_none(
|
|
if exists { Some(path.to_string()) } else { None },
|
|
"Variable",
|
|
path,
|
|
)?
|
|
};
|
|
|
|
if let Some(nlabels) = &ns.labels {
|
|
sqlx::query!(
|
|
"UPDATE variable SET labels = $1, edited_at = now(), edited_by = $4 WHERE path = $2 AND workspace_id = $3",
|
|
nlabels as &[String],
|
|
&npath,
|
|
&w_id,
|
|
&authed.username
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
}
|
|
|
|
if let Some(ws_specific) = ns.ws_specific {
|
|
if ws_specific {
|
|
sqlx::query!(
|
|
"INSERT INTO ws_specific (workspace_id, item_kind, path) VALUES ($1, 'variable', $2) ON CONFLICT DO NOTHING",
|
|
w_id,
|
|
&npath,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
} else {
|
|
sqlx::query!(
|
|
"DELETE FROM ws_specific WHERE workspace_id = $1 AND item_kind = 'variable' AND path = $2",
|
|
w_id,
|
|
&npath,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
}
|
|
}
|
|
|
|
audit_log(
|
|
&mut *tx,
|
|
&authed,
|
|
"variables.update",
|
|
ActionKind::Update,
|
|
&w_id,
|
|
Some(path),
|
|
None,
|
|
)
|
|
.await?;
|
|
|
|
// Clean up old account if it's no longer referenced and different from new account
|
|
if let Some(old_acc_id) = old_account_id {
|
|
if ns.account.is_some() && ns.account != Some(old_acc_id) {
|
|
// Check if old account is still referenced by other variables or resources
|
|
let account_still_used = sqlx::query_scalar!(
|
|
"SELECT EXISTS(SELECT 1 FROM variable WHERE account = $1 AND workspace_id = $2)",
|
|
old_acc_id,
|
|
&w_id
|
|
)
|
|
.fetch_one(&mut *tx)
|
|
.await?
|
|
.unwrap_or(true);
|
|
|
|
if !account_still_used {
|
|
// Delete the orphaned account
|
|
sqlx::query!(
|
|
"DELETE FROM account WHERE id = $1 AND workspace_id = $2",
|
|
old_acc_id,
|
|
&w_id
|
|
)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
}
|
|
}
|
|
}
|
|
|
|
tx.commit().await?;
|
|
|
|
// Detect if this was a rename operation
|
|
let old_path_if_renamed = if npath != path { Some(path) } else { None };
|
|
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Variable { path: npath.clone(), parent_path: Some(path.to_string()) },
|
|
None,
|
|
true,
|
|
old_path_if_renamed,
|
|
)
|
|
.await?;
|
|
|
|
webhook.send_message(
|
|
w_id.clone(),
|
|
WebhookMessage::UpdateVariable {
|
|
workspace: w_id,
|
|
old_path: path.to_owned(),
|
|
new_path: npath.clone(),
|
|
},
|
|
);
|
|
|
|
Ok(format!("variable {} updated (npath: {:?})", path, npath))
|
|
}
|
|
|
|
fn replace_path(v: serde_json::Value, path: &str, npath: &str) -> Value {
|
|
match v {
|
|
Value::Object(v) => Value::Object(
|
|
v.into_iter()
|
|
.map(|(k, v)| (k, replace_path(v, path, npath)))
|
|
.collect(),
|
|
),
|
|
Value::Array(arr) => Value::Array(
|
|
arr.into_iter()
|
|
.map(|v| replace_path(v, path, npath))
|
|
.collect(),
|
|
),
|
|
Value::String(s) if s == path => Value::String(npath.to_owned()),
|
|
_ => v,
|
|
}
|
|
}
|
|
|
|
/// 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<String> {
|
|
// 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);
|
|
}
|
|
}
|
|
|
|
let mut tx = db_with_opt_authed.begin().await?;
|
|
let variable_o = sqlx::query!(
|
|
"SELECT value, account, (now() > account.expires_at) as is_expired, is_secret, path from variable
|
|
LEFT JOIN account ON variable.account = account.id WHERE variable.path = $1 AND variable.workspace_id = $2", path, w_id
|
|
)
|
|
.fetch_optional(&mut *tx)
|
|
.warn_after_seconds(5)
|
|
.await?;
|
|
drop(tx);
|
|
|
|
let variable = if let Some(variable) = variable_o {
|
|
variable
|
|
} else {
|
|
explain_variable_perm_error(path, w_id, &db_with_opt_authed.db()).await?;
|
|
unreachable!()
|
|
};
|
|
|
|
let r = if variable.is_secret {
|
|
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() {
|
|
#[cfg(feature = "oauth2")]
|
|
{
|
|
let db = db_with_opt_authed.db();
|
|
let refresh_tx = db
|
|
.begin()
|
|
.await
|
|
.map_err(|e| Error::InternalErr(e.to_string()))?;
|
|
crate::oauth_refresh_oss::_refresh_token(
|
|
refresh_tx,
|
|
&variable.path,
|
|
&w_id,
|
|
variable.account.unwrap(),
|
|
db,
|
|
)
|
|
.await?
|
|
}
|
|
#[cfg(not(feature = "oauth2"))]
|
|
return Err(Error::internal_err("Require oauth2 feature".to_string()));
|
|
} else if !value.is_empty() {
|
|
// Use secret backend for decryption (supports both DB and Vault)
|
|
get_secret_value(db_with_opt_authed.db(), &w_id, &variable.path, &value).await?
|
|
} else {
|
|
"".to_string()
|
|
}
|
|
} else {
|
|
variable.value
|
|
};
|
|
|
|
if variable.is_secret && !r.is_empty() {
|
|
windmill_common::sensitive_log_masks::register_secret_for_all_running_jobs(&r);
|
|
}
|
|
|
|
// 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)
|
|
}
|