mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-21 08:02:38 +00:00
Merge remote-tracking branch 'origin/datatable-roles-redesign-part-2' into datatable-roles-redesign-part-4
# Conflicts: # backend/windmill-common/src/workspaces.rs
This commit is contained in:
@@ -1147,6 +1147,60 @@ async fn an_alias_saved_elsewhere_waits_for_roles_going_on_for_its_database(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
|
||||
async fn a_fork_waits_for_a_rename_of_the_data_table_it_keeps(
|
||||
db: Pool<Postgres>,
|
||||
) -> anyhow::Result<()> {
|
||||
initialize_tracing().await;
|
||||
// A rename in flight holds the parent's settings row and moves only the pointers it can see; a
|
||||
// fork being created is invisible to it, so the fork has to read the name the rename commits.
|
||||
let mut renaming = db.begin().await?;
|
||||
sqlx::query(
|
||||
"SELECT 1 FROM workspace_settings WHERE workspace_id = 'test-workspace' FOR UPDATE",
|
||||
)
|
||||
.execute(&mut *renaming)
|
||||
.await?;
|
||||
sqlx::query(
|
||||
"UPDATE workspace_settings SET datatable = jsonb_set(datatable #- '{datatables,main}',
|
||||
'{datatables,renamed}', datatable->'datatables'->'main')
|
||||
WHERE workspace_id = 'test-workspace'",
|
||||
)
|
||||
.execute(&mut *renaming)
|
||||
.await?;
|
||||
|
||||
let server = ApiServer::start(db.clone()).await?;
|
||||
let url = format!(
|
||||
"http://localhost:{}/api/w/test-workspace/workspaces/create_fork",
|
||||
server.addr.port()
|
||||
);
|
||||
let fork = tokio::spawn(
|
||||
authed(client().post(&url), "SECRET_TOKEN")
|
||||
.json(&json!({ "id": "wm-fork-race", "name": "race", "color": "#0000ff" }))
|
||||
.send(),
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||
assert!(
|
||||
!fork.is_finished(),
|
||||
"the fork copied the parent's data tables while a rename held them"
|
||||
);
|
||||
renaming.commit().await?;
|
||||
|
||||
let resp = fork.await??;
|
||||
assert!(resp.status().is_success(), "{}", resp.text().await?);
|
||||
let datatables: Option<Value> = sqlx::query_scalar(
|
||||
"SELECT datatable->'datatables' FROM workspace_settings WHERE workspace_id = 'wm-fork-race'",
|
||||
)
|
||||
.fetch_one(&db)
|
||||
.await?;
|
||||
let datatables = datatables.unwrap();
|
||||
assert_eq!(
|
||||
datatables["renamed"]["reference"],
|
||||
json!({ "workspace_id": "test-workspace", "datatable": "renamed" }),
|
||||
"{datatables}"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(not(all(feature = "private", feature = "enterprise")))]
|
||||
const ENTERPRISE_REFUSAL: &str = "Data table roles are a Windmill Enterprise Edition feature";
|
||||
|
||||
|
||||
@@ -2205,17 +2205,15 @@ async fn list_datatables(
|
||||
Extension(db): Extension<DB>,
|
||||
Path(w_id): Path<String>,
|
||||
) -> JsonResult<Vec<DataTableListItem>> {
|
||||
let names = list_datatable_names(&db, &w_id).await?;
|
||||
// A pointer entry owns no database, so what it resolves to is the only truthful answer here.
|
||||
// One that resolves to nothing — a pointer whose workspace was deleted — is dropped rather than
|
||||
// listed with a database it does not have; what happened is named where it is actionable
|
||||
// instead: by the delete that stranded it, and by any attempt to use it.
|
||||
let resolved =
|
||||
windmill_common::workspaces::resolve_workspace_governing_datatables(&db, &w_id).await?;
|
||||
|
||||
let mut items = Vec::with_capacity(names.len());
|
||||
for name in names {
|
||||
// A pointer entry owns no database, so what it resolves to is the only truthful answer
|
||||
// here. One that resolves to nothing — a pointer whose workspace was deleted — is dropped
|
||||
// rather than listed with a database it does not have; what happened is named where it is
|
||||
// actionable instead: by the delete that stranded it, and by any attempt to use it.
|
||||
let Ok(governing) = resolve_governing_datatable(&db, &w_id, &name).await else {
|
||||
continue;
|
||||
};
|
||||
let mut items = Vec::with_capacity(resolved.len());
|
||||
for (name, governing) in resolved {
|
||||
let database = governing
|
||||
.datatable
|
||||
.database
|
||||
@@ -8662,6 +8660,14 @@ async fn create_workspace_fork(
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
// The pointers this fork writes to the parent's data tables stay invisible until it commits, so
|
||||
// a rename of one of them cannot carry them. Holding the parent's settings row makes such a
|
||||
// rename wait for this commit, and makes the copy below read one that committed first.
|
||||
sqlx::query("SELECT 1 FROM workspace_settings WHERE workspace_id = $1 FOR SHARE")
|
||||
.bind(&parent_workspace_id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
// Clone all data from the parent workspace using Rust implementation
|
||||
if let Err(e) =
|
||||
clone_workspace_data(&mut tx, &db, &parent_workspace_id, &forked_id, &authed).await
|
||||
|
||||
@@ -233,6 +233,9 @@ pub fn role_id_by_name<'a>(catalog: &'a DatatableRoleCatalog, name: &str) -> Res
|
||||
|
||||
/// Every instance database the registry knows about. Role provisioning has to reach all of them:
|
||||
/// a role that cannot `CONNECT` to a database is refused by Postgres before any grant matters.
|
||||
///
|
||||
/// Authorization: checks nothing, and names every instance database across all workspaces. Callers
|
||||
/// MUST be superadmin-gated or keep the names server-side; never return them to a workspace caller.
|
||||
pub async fn registered_instance_databases(db: &DB) -> Result<Vec<String>> {
|
||||
crate::datatable_roles_oss::registered_instance_databases(db).await
|
||||
}
|
||||
|
||||
@@ -1522,6 +1522,96 @@ pub async fn resolve_governing_datatable(
|
||||
)))
|
||||
}
|
||||
|
||||
/// Every entry of a workspace resolved as [`resolve_governing_datatable`] resolves one, in stored
|
||||
/// order, reading the settings rows one pointer hop at a time rather than once per entry. An entry
|
||||
/// that does not resolve — malformed, a dangling pointer, a loop — is left out. Same authorization
|
||||
/// contract as the single resolution: it checks nothing.
|
||||
pub async fn resolve_workspace_governing_datatables(
|
||||
db: &DB,
|
||||
w_id: &str,
|
||||
) -> Result<Vec<(String, GoverningDatatable)>> {
|
||||
type Entries =
|
||||
std::collections::HashMap<String, std::collections::HashMap<String, serde_json::Value>>;
|
||||
async fn load(db: &DB, workspaces: &[String], entries: &mut Entries) -> Result<Vec<String>> {
|
||||
let rows: Vec<(String, String, serde_json::Value)> = sqlx::query_as(
|
||||
"SELECT ws.workspace_id, dt.key, dt.value FROM workspace_settings ws
|
||||
CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt
|
||||
WHERE ws.workspace_id = ANY($1)",
|
||||
)
|
||||
.bind(workspaces)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
for ws in workspaces {
|
||||
entries.entry(ws.clone()).or_default();
|
||||
}
|
||||
let mut keys = Vec::with_capacity(rows.len());
|
||||
for (ws, key, value) in rows {
|
||||
keys.push(key.clone());
|
||||
entries.entry(ws).or_default().insert(key, value);
|
||||
}
|
||||
Ok(keys)
|
||||
}
|
||||
|
||||
let mut entries = Entries::new();
|
||||
let listed = load(db, &[w_id.to_string()], &mut entries).await?;
|
||||
// (index into `listed`, workspace, entry name) still to be followed.
|
||||
let mut cursors: Vec<(usize, String, String)> = listed
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, name)| (i, w_id.to_string(), name.clone()))
|
||||
.collect();
|
||||
let mut resolved: Vec<(usize, GoverningDatatable)> = vec![];
|
||||
|
||||
for _ in 0..DATATABLE_REFERENCE_MAX_DEPTH {
|
||||
let mut next = vec![];
|
||||
for (i, ws, name) in cursors.drain(..) {
|
||||
let Some(value) = entries
|
||||
.get(&ws)
|
||||
.and_then(|m| m.get(&name))
|
||||
.filter(|v| !v.is_null())
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let Ok(datatable) = serde_json::from_value::<DataTable>(value.clone()) else {
|
||||
continue;
|
||||
};
|
||||
if validate_datatable_shape(&name, &datatable).is_err() {
|
||||
continue;
|
||||
}
|
||||
match &datatable.reference {
|
||||
None => {
|
||||
resolved.push((i, GoverningDatatable { workspace_id: ws, name, datatable }))
|
||||
}
|
||||
Some(reference) => next.push((
|
||||
i,
|
||||
reference.workspace_id.clone(),
|
||||
reference.datatable.clone(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
if next.is_empty() {
|
||||
break;
|
||||
}
|
||||
let to_load: Vec<String> = next
|
||||
.iter()
|
||||
.map(|(_, ws, _)| ws.clone())
|
||||
.filter(|ws| !entries.contains_key(ws))
|
||||
.collect::<std::collections::BTreeSet<_>>()
|
||||
.into_iter()
|
||||
.collect();
|
||||
if !to_load.is_empty() {
|
||||
load(db, &to_load, &mut entries).await?;
|
||||
}
|
||||
cursors = next;
|
||||
}
|
||||
|
||||
resolved.sort_by_key(|(i, _)| *i);
|
||||
Ok(resolved
|
||||
.into_iter()
|
||||
.map(|(i, governing)| (listed[i].clone(), governing))
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Build the `admin` connection for a governing entry: `custom_instance_user` for an instance
|
||||
/// database, the user's own resource for a BYO-postgres one.
|
||||
async fn resolve_datatable_connection_unchecked(
|
||||
@@ -1879,6 +1969,10 @@ pub fn strip_datatable_permissions(
|
||||
/// looked up first, so `sales?role=x` never reaches a different entry than the one stored so.
|
||||
/// When `sales` is stored too, the reference means either one, and is refused rather than
|
||||
/// resolved to whichever is looked up first.
|
||||
///
|
||||
/// Authorization: checks nothing, and its answer reveals whether `w_id` stores that exact name.
|
||||
/// Callers MUST already act for `w_id` — a job of it, or a caller authenticated into it — and
|
||||
/// MUST still pass the name to [`get_datatable_resource_from_db`] or an admin-access check.
|
||||
pub async fn parse_datatable_ref_for(
|
||||
db: &DB,
|
||||
w_id: &str,
|
||||
|
||||
@@ -377,7 +377,10 @@ pub async fn get_raw_postgres_connection(
|
||||
/// A replication stream reads every row of every table whatever the data table's roles grant, so
|
||||
/// the two don't mix: a data table under roles takes no triggers or captures, and roles cannot be
|
||||
/// turned on while one is enabled on it.
|
||||
pub async fn ensure_not_under_roles(
|
||||
///
|
||||
/// Authorization: checks nothing, and its refusal says whether `w_id`'s data table is under roles.
|
||||
/// Callers MUST have established that the caller may manage triggers in `w_id` first.
|
||||
pub(crate) async fn ensure_not_under_roles(
|
||||
db: &DB,
|
||||
w_id: &str,
|
||||
postgres_resource_path: &str,
|
||||
|
||||
@@ -2718,6 +2718,11 @@ fn pg_secret_attach_statements(db_resource: Value, alias_name: &str) -> Result<V
|
||||
esc(res.password.as_deref().unwrap_or("")),
|
||||
),
|
||||
format!("ATTACH 'sslmode={sslmode}' AS {alias_name} (TYPE postgres, SECRET {secret_name});"),
|
||||
// The attachment keeps its own resolved connection string, so the secret is dead weight
|
||||
// once attached — and a live one is a credential the script's own statements can name: an
|
||||
// `ATTACH 'dbname=<other>' (TYPE postgres, SECRET …)` would reach a database nobody
|
||||
// authorized this job for, as this role.
|
||||
format!("DROP TEMPORARY SECRET {secret_name};"),
|
||||
])
|
||||
}
|
||||
|
||||
@@ -3945,6 +3950,8 @@ mod tests {
|
||||
stmts[3],
|
||||
format!("ATTACH 'sslmode=require' AS dt (TYPE postgres, SECRET {secret_name});")
|
||||
);
|
||||
assert_eq!(stmts[4], format!("DROP TEMPORARY SECRET {secret_name};"));
|
||||
assert_eq!(stmts.len(), 5);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Reference in New Issue
Block a user