mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-21 00:02:30 +00:00
fix(datatables): give the role catalog its own row, out of reach of the config machinery
Putting it inside `custom_instance_pg_databases` was the wrong call, and it cost two ways. The catalog serializes a generated Postgres password per role, and that row is the operator-facing instance config, so the passwords reached `get_instance_config` and its YAML editor — a live cluster credential in a response body, a UI field and any log of either. Worse in the other direction: `to_settings_map` strips the catalog, so a full-row upsert of that key writes the row back without it and the catalog is gone, while the cluster keeps every login it described. `custom_instance_replication_pwd` is the precedent and says exactly why — a generated secret, written only by the server, never operator-authored, hidden so the config machinery cannot read, rewrite or drop it. The catalog is the same thing, so it now has the same shape: `datatable_roles`, in `HIDDEN_SETTINGS`, `PROTECTED_SETTINGS` and the agent-worker denylist. No redaction to keep in step with three code paths, and no way for a neighbouring write to take it out. Two races on the same shared documents. `edit_datatable_config` read the stored data tables outside its transaction and then wrote the whole `datatable` document, so a permissions save committing in between was silently rolled back; it now reads under `FOR UPDATE`. And `set_datatable_permissions` validated role ids against the catalog before opening its transaction, so a deletion in between let it write a deleted role back — including as the default, which every later job then fails on; it now holds the catalog lock and the settings row across validation and write. Completes the authorization contracts the previous commit claimed but did not finish: `read_datatable_entry` (which it named and missed), `resolve_governing_datatable`, whose whole job is to answer for a workspace the caller may not belong to, and `converge_connect_grants_with`, which had not inherited its wrapper's. Also the generic Python SDK reference: `_format_py_params` learned the bare `*` last time, but `extract_py_functions` is a second formatter and still rendered `datatable(name, role)`, so code written from that page passed a keyword-only argument positionally. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR
This commit is contained in:
co-authored by
Claude Opus 5
parent
ea14c01a3c
commit
715d8a0e6d
+22
@@ -0,0 +1,22 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT 1 AS one FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "one",
|
||||
"type_info": "Int4"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "06abbf945bee93349ff88f64906b96ea1e853ef202510281427cfa9beeff81b3"
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO global_settings (name, value) VALUES ($1, $2)\n ON CONFLICT (name) DO UPDATE SET value = $2",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Varchar",
|
||||
"Jsonb"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "6f9fb5d72f486358fa25d6887bd69b93910e028f140c07048f2c1c8d63ee6909"
|
||||
}
|
||||
+22
@@ -0,0 +1,22 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT ws.datatable->'datatables' FROM workspace_settings ws\n WHERE ws.workspace_id = $1 FOR UPDATE",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "?column?",
|
||||
"type_info": "Jsonb"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "b42af37fb474bea4c5419b0a46d9eadfe384013ab970ccf9c5effd1c78321b7c"
|
||||
}
|
||||
@@ -354,15 +354,7 @@ async fn concurrent_role_catalog_writes_do_not_lose_an_entry(db: Pool<Postgres>)
|
||||
id.to_string(),
|
||||
InstanceDatatableRole { name: id.to_string(), enabled: true, pwd: None },
|
||||
);
|
||||
let value = serde_json::to_value(&catalog)?;
|
||||
sqlx::query(
|
||||
"UPDATE global_settings
|
||||
SET value = jsonb_set(COALESCE(value, '{}'::jsonb), '{roles}', $1)
|
||||
WHERE name = 'custom_instance_pg_databases'",
|
||||
)
|
||||
.bind(value)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
windmill_common::datatable_roles::write_role_catalog(&mut tx, &catalog).await?;
|
||||
tx.commit().await?;
|
||||
Ok::<_, anyhow::Error>(())
|
||||
}
|
||||
|
||||
@@ -2,14 +2,12 @@
|
||||
-- carrying a copy. `test-user-2` is a non-admin of the parent and an admin of the fork: the shape
|
||||
-- the pointer exists for.
|
||||
|
||||
UPDATE global_settings SET value = jsonb_set(value, '{roles}',
|
||||
'{"role1": {"name": "analytics", "enabled": true, "pwd": "pw"}}'::jsonb)
|
||||
WHERE name = 'custom_instance_pg_databases';
|
||||
INSERT INTO global_settings (name, value)
|
||||
SELECT 'custom_instance_pg_databases',
|
||||
'{"user_pwd": "pw", "databases": {"dt_main": {}},
|
||||
"roles": {"role1": {"name": "analytics", "enabled": true, "pwd": "pw"}}}'::jsonb
|
||||
WHERE NOT EXISTS (SELECT 1 FROM global_settings WHERE name = 'custom_instance_pg_databases');
|
||||
INSERT INTO global_settings (name, value) VALUES
|
||||
('custom_instance_pg_databases', '{"user_pwd": "pw", "databases": {"dt_main": {}}}'::jsonb),
|
||||
-- The role catalog has its own row: it holds generated credentials and must stay out of the
|
||||
-- operator-facing config the neighbouring row belongs to.
|
||||
('datatable_roles', '{"role1": {"name": "analytics", "enabled": true, "pwd": "pw"}}'::jsonb)
|
||||
ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value;
|
||||
|
||||
UPDATE workspace_settings SET datatable = '{
|
||||
"datatables": {
|
||||
|
||||
@@ -2633,40 +2633,6 @@ fn datatable_role_infos(
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Persist the catalog next to the instance Postgres password, in the same `global_settings` row
|
||||
/// the instance database registry lives in.
|
||||
///
|
||||
/// Errors when it matches nothing. The cluster is written first, so a silent no-op here would
|
||||
/// leave a live Postgres login with a password nobody recorded: invisible to the catalog,
|
||||
/// un-recreatable (the name is taken) and un-deletable (there is no entry to delete). The row is
|
||||
/// normally planted by the boot converge, but that swallows its own failures, so this is a check
|
||||
/// rather than an assumption.
|
||||
async fn write_role_catalog(
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
catalog: &windmill_common::datatable_roles::DatatableRoleCatalog,
|
||||
) -> error::Result<()> {
|
||||
let value = serde_json::to_value(catalog).map_err(to_anyhow)?;
|
||||
let written = sqlx::query!(
|
||||
"UPDATE global_settings SET value = jsonb_set(COALESCE(value, '{}'::jsonb), '{roles}', $1)
|
||||
WHERE name = 'custom_instance_pg_databases'",
|
||||
value
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.await?
|
||||
.rows_affected();
|
||||
if written == 0 {
|
||||
return Err(error::Error::internal_err(
|
||||
concat!(
|
||||
"The instance Postgres settings row is missing, so the data table role catalog ",
|
||||
"could not be recorded. Refresh the custom instance user password in instance ",
|
||||
"settings to recreate it, then try again."
|
||||
)
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list_datatable_roles(
|
||||
authed: ApiAuthed,
|
||||
Extension(db): Extension<DB>,
|
||||
@@ -2711,7 +2677,7 @@ async fn create_datatable_role(
|
||||
pwd: Some(pwd),
|
||||
},
|
||||
);
|
||||
write_role_catalog(&mut tx, &catalog).await?;
|
||||
windmill_common::datatable_roles::write_role_catalog(&mut tx, &catalog).await?;
|
||||
tx.commit().await?;
|
||||
converge_connect_grants_everywhere(&db, &catalog).await;
|
||||
windmill_common::feature_usage::log_feature_usage("datatable", "role_created", "");
|
||||
@@ -2768,7 +2734,7 @@ async fn update_datatable_role(
|
||||
}
|
||||
|
||||
catalog.insert(id.clone(), updated.clone());
|
||||
write_role_catalog(&mut tx, &catalog).await?;
|
||||
windmill_common::datatable_roles::write_role_catalog(&mut tx, &catalog).await?;
|
||||
tx.commit().await?;
|
||||
converge_connect_grants_everywhere(&db, &catalog).await;
|
||||
|
||||
@@ -2811,7 +2777,7 @@ async fn delete_datatable_role(
|
||||
|
||||
windmill_common::datatable_roles::drop_instance_role(&db, &mut tx, &role.name).await?;
|
||||
catalog.remove(&id);
|
||||
write_role_catalog(&mut tx, &catalog).await?;
|
||||
windmill_common::datatable_roles::write_role_catalog(&mut tx, &catalog).await?;
|
||||
tx.commit().await?;
|
||||
// After the drop commits: a tenant naming a role that still exists is harmless, one naming a
|
||||
// role that is gone is not, so this only ever runs once the cluster agrees it is gone.
|
||||
|
||||
@@ -274,8 +274,21 @@ async fn set_datatable_permissions(
|
||||
)));
|
||||
}
|
||||
|
||||
// One transaction for the whole save, holding both locks the decision depends on: the role
|
||||
// catalog, so a role cannot be deleted between validating an id and writing it back, and the
|
||||
// workspace settings row, so a concurrent settings save cannot carry a stale copy of this
|
||||
// block forward over what is written here.
|
||||
let mut tx = db.begin().await?;
|
||||
windmill_common::datatable_roles::lock_role_catalog(&mut tx).await?;
|
||||
sqlx::query!(
|
||||
"SELECT 1 AS one FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE",
|
||||
&governing.workspace_id
|
||||
)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
|
||||
let permissions = if req.permissioned {
|
||||
let catalog = read_role_catalog(&db).await?;
|
||||
let catalog = windmill_common::datatable_roles::read_role_catalog_tx(&mut tx).await?;
|
||||
let mut roles: BTreeMap<String, DataTableRoleTenants> = BTreeMap::new();
|
||||
for role in req.roles {
|
||||
if role.id != ADMIN_DATATABLE_ROLE && !catalog.contains_key(&role.id) {
|
||||
@@ -306,30 +319,6 @@ async fn set_datatable_permissions(
|
||||
None
|
||||
};
|
||||
|
||||
// An instance database provisioned before data table roles existed has neither the grant
|
||||
// options the admin connection needs to delegate privileges, nor a CONNECT grant for any role
|
||||
// — so a role would be refused at login however its tenants read. Repair it here, at the one
|
||||
// moment someone is deciding this data table's roles. Best-effort: neither is worth failing a
|
||||
// tenant edit over, and both converge again on the next save.
|
||||
if permissions.is_some() {
|
||||
if let Some(database) = governing.datatable.database.as_ref() {
|
||||
if database.resource_type == DataTableCatalogResourceType::Instance {
|
||||
let dbname = &database.resource_path;
|
||||
if let Err(e) =
|
||||
windmill_common::ensure_instance_db_grant_options_unchecked(&db, dbname).await
|
||||
{
|
||||
tracing::warn!("Could not refresh grant options on '{dbname}': {e}");
|
||||
}
|
||||
if let Err(e) =
|
||||
windmill_common::datatable_roles::converge_connect_grants(&db, dbname).await
|
||||
{
|
||||
tracing::warn!("Could not refresh CONNECT grants on '{dbname}': {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut tx = db.begin().await?;
|
||||
let value = match &permissions {
|
||||
Some(p) => serde_json::to_value(p).map_err(|e| Error::internal_err(e.to_string()))?,
|
||||
None => serde_json::Value::Null,
|
||||
@@ -362,6 +351,32 @@ async fn set_datatable_permissions(
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
|
||||
// An instance database provisioned before data table roles existed has neither the grant
|
||||
// options the admin connection needs to delegate privileges, nor a CONNECT grant for any role
|
||||
// — so a role would be refused at login however its tenants read. Repair it here, at the one
|
||||
// moment someone is deciding this data table's roles. Best-effort: neither is worth failing a
|
||||
// tenant edit over, and both converge again on the next save.
|
||||
//
|
||||
// Runs after the commit: it opens its own connections to other databases, which has no place
|
||||
// inside a transaction holding two locks.
|
||||
if permissions.is_some() {
|
||||
if let Some(database) = governing.datatable.database.as_ref() {
|
||||
if database.resource_type == DataTableCatalogResourceType::Instance {
|
||||
let dbname = &database.resource_path;
|
||||
if let Err(e) =
|
||||
windmill_common::ensure_instance_db_grant_options_unchecked(&db, dbname).await
|
||||
{
|
||||
tracing::warn!("Could not refresh grant options on '{dbname}': {e}");
|
||||
}
|
||||
if let Err(e) =
|
||||
windmill_common::datatable_roles::converge_connect_grants(&db, dbname).await
|
||||
{
|
||||
tracing::warn!("Could not refresh CONNECT grants on '{dbname}': {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A live replication stream holds a connection it opened under the old decision. Bouncing the
|
||||
// rows makes every listener reconnect and re-authorize.
|
||||
restart_streams_reaching(&db, &governing).await?;
|
||||
|
||||
@@ -3599,12 +3599,17 @@ async fn edit_datatable_config(
|
||||
|
||||
let mut tx = db.begin().await?;
|
||||
|
||||
// Read under the row lock this transaction will write with. `permissions`, `reference` and
|
||||
// `forked_from` are carried across from what this read returns, so a permissions save
|
||||
// committing between the read and the whole-document write below would be silently rolled back
|
||||
// by it.
|
||||
let old_datatables: HashMap<String, DataTable> = serde_json::from_value(
|
||||
sqlx::query_scalar!(
|
||||
"SELECT ws.datatable->'datatables' FROM workspace_settings ws WHERE ws.workspace_id = $1",
|
||||
"SELECT ws.datatable->'datatables' FROM workspace_settings ws
|
||||
WHERE ws.workspace_id = $1 FOR UPDATE",
|
||||
&w_id
|
||||
)
|
||||
.fetch_one(&db)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?
|
||||
.unwrap_or(serde_json::Value::Null),
|
||||
)
|
||||
|
||||
@@ -21,6 +21,7 @@ use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::{
|
||||
error::{Error, Result},
|
||||
global_settings::DATATABLE_ROLES_SETTING,
|
||||
DB,
|
||||
};
|
||||
|
||||
@@ -33,9 +34,8 @@ pub const ADMIN_DATATABLE_ROLE: &str = "admin";
|
||||
/// membership is what later lets it `ALTER ... OWNER TO` a role and drop it.
|
||||
pub const CUSTOM_INSTANCE_USER: &str = "custom_instance_user";
|
||||
|
||||
/// One catalog entry. The password is per role and instance-wide; it lives here rather than in any
|
||||
/// workspace's settings, next to the `custom_instance_user` password in the same
|
||||
/// `custom_instance_pg_databases` row.
|
||||
/// One catalog entry. The password is per role and instance-wide, and lives in the instance's own
|
||||
/// [`DATATABLE_ROLES_SETTING`] row rather than in any workspace's settings.
|
||||
#[derive(Deserialize, Serialize, Clone)]
|
||||
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
|
||||
pub struct InstanceDatatableRole {
|
||||
@@ -134,11 +134,11 @@ pub async fn lock_role_catalog(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>) -
|
||||
/// response, a log line or an audit record.
|
||||
pub async fn read_role_catalog(db: &DB) -> Result<DatatableRoleCatalog> {
|
||||
let value = sqlx::query_scalar!(
|
||||
"SELECT value->'roles' FROM global_settings WHERE name = 'custom_instance_pg_databases'"
|
||||
"SELECT value FROM global_settings WHERE name = $1",
|
||||
DATATABLE_ROLES_SETTING
|
||||
)
|
||||
.fetch_optional(db)
|
||||
.await?
|
||||
.flatten();
|
||||
.await?;
|
||||
Ok(parse_role_catalog(value))
|
||||
}
|
||||
|
||||
@@ -148,14 +148,36 @@ pub async fn read_role_catalog_tx(
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
) -> Result<DatatableRoleCatalog> {
|
||||
let value = sqlx::query_scalar!(
|
||||
"SELECT value->'roles' FROM global_settings WHERE name = 'custom_instance_pg_databases'"
|
||||
"SELECT value FROM global_settings WHERE name = $1",
|
||||
DATATABLE_ROLES_SETTING
|
||||
)
|
||||
.fetch_optional(&mut **tx)
|
||||
.await?
|
||||
.flatten();
|
||||
.await?;
|
||||
Ok(parse_role_catalog(value))
|
||||
}
|
||||
|
||||
/// Persist the catalog, in the caller's transaction so it commits with the cluster DDL it
|
||||
/// describes. Upserts: the row does not exist until the first role is created.
|
||||
///
|
||||
/// Authorization: writes generated Postgres credentials. Callers MUST restrict this to superadmin
|
||||
/// paths and MUST hold [`lock_role_catalog`] on `tx`.
|
||||
pub async fn write_role_catalog(
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
catalog: &DatatableRoleCatalog,
|
||||
) -> Result<()> {
|
||||
let value = serde_json::to_value(catalog)
|
||||
.map_err(|e| Error::internal_err(format!("serializing the role catalog: {e}")))?;
|
||||
sqlx::query!(
|
||||
"INSERT INTO global_settings (name, value) VALUES ($1, $2)
|
||||
ON CONFLICT (name) DO UPDATE SET value = $2",
|
||||
DATATABLE_ROLES_SETTING,
|
||||
value
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// A catalog that will not deserialize is an empty one, which fails closed: tenants are keyed by
|
||||
/// id independently of it, so every role then resolves to "no longer exists on this instance"
|
||||
/// rather than to admin.
|
||||
@@ -212,6 +234,8 @@ pub async fn converge_connect_grants(db: &DB, dbname: &str) -> Result<()> {
|
||||
converge_connect_grants_with(db, dbname, &catalog).await
|
||||
}
|
||||
|
||||
/// As [`converge_connect_grants`], with a catalog the caller already read. Same authorization
|
||||
/// contract: it rewrites a database's ACL with the server's own credentials and checks nothing.
|
||||
pub async fn converge_connect_grants_with(
|
||||
db: &DB,
|
||||
dbname: &str,
|
||||
|
||||
@@ -355,8 +355,18 @@ pub const AGENT_WORKER_BLOCKED_SETTINGS: &[&str] = &[
|
||||
// resolve datatable connections through the dedicated datatable endpoints, never these.
|
||||
"custom_instance_pg_databases",
|
||||
"custom_instance_replication_pwd",
|
||||
// The data table role catalog: one generated Postgres password per role.
|
||||
DATATABLE_ROLES_SETTING,
|
||||
];
|
||||
|
||||
/// The instance's data table role catalog, `{ "<id>": { name, enabled, pwd } }`.
|
||||
///
|
||||
/// Its own row rather than a field of `custom_instance_pg_databases`, for the same reason
|
||||
/// `custom_instance_replication_pwd` is: it holds generated credentials and is written only by the
|
||||
/// server, so the config machinery must not be able to read it into an export, rewrite it, or drop
|
||||
/// it on a full-row upsert of a neighbour.
|
||||
pub const DATATABLE_ROLES_SETTING: &str = "datatable_roles";
|
||||
|
||||
/// Whether an agent worker may read the given global setting over HTTP.
|
||||
/// Deny-by-exception: everything is readable except [`AGENT_WORKER_BLOCKED_SETTINGS`].
|
||||
pub fn is_setting_readable_by_agent_worker(name: &str) -> bool {
|
||||
@@ -988,6 +998,7 @@ mod tests {
|
||||
OTEL_TRACING_PROXY_SETTING,
|
||||
"custom_instance_pg_databases",
|
||||
"custom_instance_replication_pwd",
|
||||
DATATABLE_ROLES_SETTING,
|
||||
] {
|
||||
assert!(
|
||||
!is_setting_readable_by_agent_worker(key),
|
||||
|
||||
@@ -440,13 +440,11 @@ impl GlobalSettings {
|
||||
serde_json::Value::Object(map) => map.into_iter().collect(),
|
||||
_ => unreachable!(),
|
||||
};
|
||||
// Strip the runtime-only sub-fields of custom_instance_pg_databases: `databases` is setup
|
||||
// status/logs managed by the setup endpoint, and `roles` is the data table role catalog,
|
||||
// which carries one Postgres password per role. Neither is configuration.
|
||||
// Strip runtime-only `databases` sub-field from custom_instance_pg_databases.
|
||||
// It contains setup status/logs managed by the setup endpoint, not configuration.
|
||||
if let Some(pg) = map.get_mut("custom_instance_pg_databases") {
|
||||
if let Some(obj) = pg.as_object_mut() {
|
||||
obj.remove("databases");
|
||||
obj.remove("roles");
|
||||
}
|
||||
}
|
||||
map
|
||||
@@ -797,10 +795,6 @@ pub struct CustomInstancePgDatabases {
|
||||
pub user_pwd: Option<StringOrSecretRef>,
|
||||
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
|
||||
pub databases: BTreeMap<String, CustomInstanceDb>,
|
||||
/// The instance's data table role catalog, keyed by generated id. Runtime state carrying one
|
||||
/// password per role, so it is stripped from config sync exactly like `databases`.
|
||||
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
|
||||
pub roles: BTreeMap<String, crate::datatable_roles::InstanceDatatableRole>,
|
||||
}
|
||||
|
||||
/// Status of a single custom instance database.
|
||||
@@ -975,6 +969,7 @@ pub const PROTECTED_SETTINGS: &[&str] = &[
|
||||
"ducklake_settings",
|
||||
"custom_instance_pg_databases",
|
||||
"custom_instance_replication_pwd",
|
||||
crate::global_settings::DATATABLE_ROLES_SETTING,
|
||||
"uid",
|
||||
"rsa_keys",
|
||||
"jwt_secret",
|
||||
@@ -1000,6 +995,9 @@ pub const HIDDEN_SETTINGS: &[&str] = &[
|
||||
// Server-only (written by setup/refresh via direct SQL), never operator-authored —
|
||||
// hidden so the config machinery can't read, rewrite, or drop it.
|
||||
"custom_instance_replication_pwd",
|
||||
// The data table role catalog, one generated Postgres password per role. Same reasoning as
|
||||
// the line above: server-written, never operator-authored, and it must not reach an export.
|
||||
crate::global_settings::DATATABLE_ROLES_SETTING,
|
||||
];
|
||||
|
||||
/// Top-level settings whose entire value is sensitive and must be fully redacted in logs.
|
||||
@@ -1211,17 +1209,14 @@ pub fn diff_global_settings(
|
||||
} else {
|
||||
desired_value.clone()
|
||||
};
|
||||
// Preserve the runtime-only sub-fields inside `custom_instance_pg_databases` so that
|
||||
// config sync never wipes setup status/logs managed by the setup endpoint, nor the data
|
||||
// table role catalog — the latter mirrors real Postgres roles, so losing it would leave
|
||||
// the cluster holding logins Windmill can no longer name.
|
||||
// Preserve the runtime-only `databases` sub-field inside
|
||||
// `custom_instance_pg_databases` so that config sync never wipes
|
||||
// setup status/logs that are managed by the setup endpoint.
|
||||
if key == "custom_instance_pg_databases" {
|
||||
if let Some(existing) = current.get(key) {
|
||||
for runtime_field in ["databases", "roles"] {
|
||||
if let Some(kept) = existing.get(runtime_field) {
|
||||
if let Some(obj) = value.as_object_mut() {
|
||||
obj.entry(runtime_field).or_insert_with(|| kept.clone());
|
||||
}
|
||||
if let Some(databases) = existing.get("databases") {
|
||||
if let Some(obj) = value.as_object_mut() {
|
||||
obj.entry("databases").or_insert_with(|| databases.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1426,6 +1426,11 @@ fn datatable_not_found_error(name: &str, datatables: Option<&serde_json::Value>)
|
||||
}
|
||||
|
||||
/// Read one workspace's data table entry, without following a pointer.
|
||||
///
|
||||
/// Authorization: reads a workspace's stored configuration by id and checks nothing — not that the
|
||||
/// caller belongs to that workspace, nor that they may see the data table. Callers MUST have
|
||||
/// authorized access to `w_id` already, and MUST NOT return the entry to a caller from another
|
||||
/// workspace: it names the database and, on a governing entry, who may reach it as what.
|
||||
pub async fn read_datatable_entry(db: &DB, w_id: &str, name: &str) -> Result<DataTable> {
|
||||
let datatables = sqlx::query_scalar!(
|
||||
r#"
|
||||
@@ -1453,6 +1458,13 @@ pub async fn read_datatable_entry(db: &DB, w_id: &str, name: &str) -> Result<Dat
|
||||
/// Every decision downstream — which database to connect to, whose `permissions` apply, whose
|
||||
/// members tenants are evaluated against, who may administer it — is taken on this, never on the
|
||||
/// entry the caller named.
|
||||
///
|
||||
/// Authorization: resolving deliberately crosses into the governing workspace, so it answers for a
|
||||
/// workspace the caller may not belong to and checks nothing itself. It is the input to the
|
||||
/// checks, not one of them: callers MUST pass what it returns to
|
||||
/// [`can_use_datatable_role_in_governing_workspace`] or [`ensure_datatable_admin_access`] before
|
||||
/// acting on it, and MUST NOT return its `permissions` or `workspace_id` to a caller from
|
||||
/// elsewhere without gating on the answer.
|
||||
pub struct GoverningDatatable {
|
||||
pub workspace_id: String,
|
||||
pub name: String,
|
||||
|
||||
Generated
+5
-5
@@ -4426,7 +4426,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
|
||||
#
|
||||
# Returns:
|
||||
# DataTableClient instance
|
||||
def datatable(name: str = 'main', role: Optional[str] = None)
|
||||
def datatable(name: str = 'main', *, role: Optional[str] = None)
|
||||
|
||||
# Get a DuckLake client for DuckDB queries.
|
||||
#
|
||||
@@ -4644,7 +4644,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
|
||||
#
|
||||
# @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
|
||||
# async def call_api(payload: dict): ...
|
||||
def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill script.
|
||||
#
|
||||
@@ -4657,7 +4657,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
|
||||
# @workflow
|
||||
# async def main():
|
||||
# data = await extract(url="https://...")
|
||||
def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill flow.
|
||||
#
|
||||
@@ -4670,7 +4670,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
|
||||
# @workflow
|
||||
# async def main():
|
||||
# result = await pipeline(input=data)
|
||||
def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Decorator marking an async function as a workflow-as-code entry point.
|
||||
#
|
||||
@@ -4735,7 +4735,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
|
||||
# ...
|
||||
#
|
||||
# results = await parallel(items, process, concurrency=5)
|
||||
async def parallel(items, fn, concurrency: Optional[int] = None)
|
||||
async def parallel(items, fn, *, concurrency: Optional[int] = None)
|
||||
|
||||
# Commit Kafka offsets for a trigger with auto_commit disabled.
|
||||
#
|
||||
|
||||
@@ -2396,7 +2396,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
|
||||
#
|
||||
# Returns:
|
||||
# DataTableClient instance
|
||||
def datatable(name: str = 'main', role: Optional[str] = None)
|
||||
def datatable(name: str = 'main', *, role: Optional[str] = None)
|
||||
|
||||
# Get a DuckLake client for DuckDB queries.
|
||||
#
|
||||
@@ -2614,7 +2614,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
|
||||
#
|
||||
# @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
|
||||
# async def call_api(payload: dict): ...
|
||||
def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill script.
|
||||
#
|
||||
@@ -2627,7 +2627,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
|
||||
# @workflow
|
||||
# async def main():
|
||||
# data = await extract(url="https://...")
|
||||
def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill flow.
|
||||
#
|
||||
@@ -2640,7 +2640,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
|
||||
# @workflow
|
||||
# async def main():
|
||||
# result = await pipeline(input=data)
|
||||
def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Decorator marking an async function as a workflow-as-code entry point.
|
||||
#
|
||||
@@ -2705,7 +2705,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
|
||||
# ...
|
||||
#
|
||||
# results = await parallel(items, process, concurrency=5)
|
||||
async def parallel(items, fn, concurrency: Optional[int] = None)
|
||||
async def parallel(items, fn, *, concurrency: Optional[int] = None)
|
||||
|
||||
# Commit Kafka offsets for a trigger with auto_commit disabled.
|
||||
#
|
||||
|
||||
@@ -2536,7 +2536,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
|
||||
#
|
||||
# Returns:
|
||||
# DataTableClient instance
|
||||
def datatable(name: str = 'main', role: Optional[str] = None)
|
||||
def datatable(name: str = 'main', *, role: Optional[str] = None)
|
||||
|
||||
# Get a DuckLake client for DuckDB queries.
|
||||
#
|
||||
@@ -2754,7 +2754,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
|
||||
#
|
||||
# @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
|
||||
# async def call_api(payload: dict): ...
|
||||
def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill script.
|
||||
#
|
||||
@@ -2767,7 +2767,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
|
||||
# @workflow
|
||||
# async def main():
|
||||
# data = await extract(url="https://...")
|
||||
def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill flow.
|
||||
#
|
||||
@@ -2780,7 +2780,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
|
||||
# @workflow
|
||||
# async def main():
|
||||
# result = await pipeline(input=data)
|
||||
def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Decorator marking an async function as a workflow-as-code entry point.
|
||||
#
|
||||
@@ -2845,7 +2845,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
|
||||
# ...
|
||||
#
|
||||
# results = await parallel(items, process, concurrency=5)
|
||||
async def parallel(items, fn, concurrency: Optional[int] = None)
|
||||
async def parallel(items, fn, *, concurrency: Optional[int] = None)
|
||||
|
||||
# Commit Kafka offsets for a trigger with auto_commit disabled.
|
||||
#
|
||||
|
||||
@@ -475,7 +475,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
|
||||
#
|
||||
# Returns:
|
||||
# DataTableClient instance
|
||||
def datatable(name: str = 'main', role: Optional[str] = None)
|
||||
def datatable(name: str = 'main', *, role: Optional[str] = None)
|
||||
|
||||
# Get a DuckLake client for DuckDB queries.
|
||||
#
|
||||
@@ -693,7 +693,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
|
||||
#
|
||||
# @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
|
||||
# async def call_api(payload: dict): ...
|
||||
def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill script.
|
||||
#
|
||||
@@ -706,7 +706,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
|
||||
# @workflow
|
||||
# async def main():
|
||||
# data = await extract(url="https://...")
|
||||
def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill flow.
|
||||
#
|
||||
@@ -719,7 +719,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
|
||||
# @workflow
|
||||
# async def main():
|
||||
# result = await pipeline(input=data)
|
||||
def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Decorator marking an async function as a workflow-as-code entry point.
|
||||
#
|
||||
@@ -784,7 +784,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
|
||||
# ...
|
||||
#
|
||||
# results = await parallel(items, process, concurrency=5)
|
||||
async def parallel(items, fn, concurrency: Optional[int] = None)
|
||||
async def parallel(items, fn, *, concurrency: Optional[int] = None)
|
||||
|
||||
# Commit Kafka offsets for a trigger with auto_commit disabled.
|
||||
#
|
||||
|
||||
@@ -660,7 +660,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
|
||||
#
|
||||
# Returns:
|
||||
# DataTableClient instance
|
||||
def datatable(name: str = 'main', role: Optional[str] = None)
|
||||
def datatable(name: str = 'main', *, role: Optional[str] = None)
|
||||
|
||||
# Get a DuckLake client for DuckDB queries.
|
||||
#
|
||||
@@ -878,7 +878,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
|
||||
#
|
||||
# @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
|
||||
# async def call_api(payload: dict): ...
|
||||
def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill script.
|
||||
#
|
||||
@@ -891,7 +891,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
|
||||
# @workflow
|
||||
# async def main():
|
||||
# data = await extract(url="https://...")
|
||||
def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Create a task that dispatches to a separate Windmill flow.
|
||||
#
|
||||
@@ -904,7 +904,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
|
||||
# @workflow
|
||||
# async def main():
|
||||
# result = await pipeline(input=data)
|
||||
def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
|
||||
|
||||
# Decorator marking an async function as a workflow-as-code entry point.
|
||||
#
|
||||
@@ -969,7 +969,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
|
||||
# ...
|
||||
#
|
||||
# results = await parallel(items, process, concurrency=5)
|
||||
async def parallel(items, fn, concurrency: Optional[int] = None)
|
||||
async def parallel(items, fn, *, concurrency: Optional[int] = None)
|
||||
|
||||
# Commit Kafka offsets for a trigger with auto_commit disabled.
|
||||
#
|
||||
|
||||
@@ -223,7 +223,12 @@ def extract_py_functions(content: str) -> list[dict]:
|
||||
if args.vararg:
|
||||
params.append(f"*{args.vararg.arg}")
|
||||
|
||||
# Handle keyword-only args
|
||||
# Handle keyword-only args. Same bare `*` as `_format_py_params`: without it the rendered
|
||||
# signature reads as all-positional, and code written against this reference passes a
|
||||
# keyword-only argument positionally and gets a TypeError.
|
||||
if args.kwonlyargs and not args.vararg:
|
||||
params.append('*')
|
||||
|
||||
for i, arg in enumerate(args.kwonlyargs):
|
||||
param_str = arg.arg
|
||||
if arg.annotation:
|
||||
|
||||
Reference in New Issue
Block a user