mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-17 16:02:25 +00:00
Compare commits
23
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
48f026eb0f | ||
|
|
0e659898f3 | ||
|
|
a296ca016a | ||
|
|
7c16d482fd | ||
|
|
c55ff3141c | ||
|
|
5eec203099 | ||
|
|
0b3882f2b2 | ||
|
|
58731b1828 | ||
|
|
53d62dd399 | ||
|
|
e17b76c255 | ||
|
|
5ddaee5b78 | ||
|
|
0406f133c5 | ||
|
|
b62d8b246a | ||
|
|
3981b2eb1b | ||
|
|
5bc2f13565 | ||
|
|
0c53990d18 | ||
|
|
9b0046d867 | ||
|
|
1f78ce77f5 | ||
|
|
67f488c676 | ||
|
|
ced4913b33 | ||
|
|
955aadbf10 | ||
|
|
a41644c795 | ||
|
|
455e1b8d23 |
-38
@@ -1,38 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT id, name, enabled, pwd FROM datatable_role",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "id",
|
||||
"type_info": "Varchar"
|
||||
},
|
||||
{
|
||||
"ordinal": 1,
|
||||
"name": "name",
|
||||
"type_info": "Varchar"
|
||||
},
|
||||
{
|
||||
"ordinal": 2,
|
||||
"name": "enabled",
|
||||
"type_info": "Bool"
|
||||
},
|
||||
{
|
||||
"ordinal": 3,
|
||||
"name": "pwd",
|
||||
"type_info": "Text"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": []
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "71ee2cb6661cca1fa4d8874a7f6d368347c59f36fd87df6dc7996152ccb84af0"
|
||||
}
|
||||
-29
@@ -1,29 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "\n SELECT ws.workspace_id AS \"workspace_id!\", dt.key AS \"datatable!\"\n FROM workspace_settings ws\n JOIN workspace w ON w.id = ws.workspace_id AND w.deleted = false\n CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n WHERE ws.workspace_id <> $1\n AND dt.value->'database'->>'resource_type' = 'instance'\n AND dt.value->'database'->>'resource_path' = $2\n ORDER BY ws.workspace_id, dt.key\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "workspace_id!",
|
||||
"type_info": "Varchar"
|
||||
},
|
||||
{
|
||||
"ordinal": 1,
|
||||
"name": "datatable!",
|
||||
"type_info": "Text"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "79799b5a2e499df6c28e286c42b9ad2db940c2455ab19cc95e5198baf96d5629"
|
||||
}
|
||||
-17
@@ -1,17 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO datatable_role (id, name, enabled, pwd) VALUES ($1, $2, $3, $4)",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Varchar",
|
||||
"Varchar",
|
||||
"Bool",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "86af9d51a158ea5cb6161461ecddf2a63695f8cbf8af648da5a0a77a5b9d02ba"
|
||||
}
|
||||
-20
@@ -1,20 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT jsonb_object_keys(value->'databases') FROM global_settings\n WHERE name = 'custom_instance_pg_databases'",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "jsonb_object_keys",
|
||||
"type_info": "Text"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": []
|
||||
},
|
||||
"nullable": [
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "b9842d2d8abf382bd82d8fa1de012373638be391f884f81dc387ffc465badac6"
|
||||
}
|
||||
-24
@@ -1,24 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT dt.key AS \"datatable!\"\n FROM workspace_settings ws\n CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt\n WHERE ws.workspace_id = $1\n AND dt.key <> $2\n AND NOT dt.value ? 'permissions'\n AND dt.value->'database'->>'resource_type' = 'instance'\n AND dt.value->'database'->>'resource_path' = $3\n ORDER BY dt.key",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "datatable!",
|
||||
"type_info": "Text"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "d48ca62c86b1af7a9dd2450c1c28dc45020a2a553d8874c49f9eafedea5a9d40"
|
||||
}
|
||||
@@ -1 +1 @@
|
||||
2de95863062cc0b00933afd86800aaf953af47d5
|
||||
50ef80045ddb208ee1feee2d9670210f703620bf
|
||||
@@ -0,0 +1,22 @@
|
||||
-- Roles on the external cluster are live logins there; dropping the column would forget them.
|
||||
LOCK TABLE datatable_role;
|
||||
DO $$
|
||||
BEGIN
|
||||
IF EXISTS (SELECT 1 FROM datatable_role WHERE cluster <> 'instance') THEN
|
||||
RAISE EXCEPTION 'datatable_role holds roles on the external instance cluster. Delete them in instance settings first.';
|
||||
END IF;
|
||||
-- Before this, only data tables on Windmill's own cluster could be under roles, and a role
|
||||
-- block left with just `admin` survives deleting every external role.
|
||||
IF EXISTS (
|
||||
SELECT 1 FROM workspace_settings ws,
|
||||
jsonb_each(CASE WHEN jsonb_typeof(ws.datatable->'datatables') = 'object'
|
||||
THEN ws.datatable->'datatables' ELSE '{}'::jsonb END) dt
|
||||
WHERE dt.value->'database'->>'resource_type' = 'external_instance'
|
||||
AND dt.value ? 'permissions'
|
||||
) THEN
|
||||
RAISE EXCEPTION 'external instance data tables are still under roles. Turn their roles off first.';
|
||||
END IF;
|
||||
END $$;
|
||||
ALTER TABLE datatable_role DROP CONSTRAINT datatable_role_cluster_name_key;
|
||||
ALTER TABLE datatable_role ADD CONSTRAINT datatable_role_name_key UNIQUE (name);
|
||||
ALTER TABLE datatable_role DROP COLUMN cluster;
|
||||
@@ -0,0 +1,8 @@
|
||||
-- A data table role is a Postgres login on one cluster: Windmill's own ('instance'), or the external
|
||||
-- instance cluster ('external_instance'). Role names are the cluster's own key, so they are unique
|
||||
-- per cluster rather than across the instance.
|
||||
ALTER TABLE datatable_role
|
||||
ADD COLUMN cluster VARCHAR(20) NOT NULL DEFAULT 'instance'
|
||||
CHECK (cluster IN ('instance', 'external_instance'));
|
||||
ALTER TABLE datatable_role DROP CONSTRAINT datatable_role_name_key;
|
||||
ALTER TABLE datatable_role ADD CONSTRAINT datatable_role_cluster_name_key UNIQUE (cluster, name);
|
||||
@@ -391,7 +391,11 @@ async fn concurrent_role_creations_both_survive(db: Pool<Postgres>) -> anyhow::R
|
||||
assert_eq!(a.0, 200, "{}", a.1);
|
||||
assert_eq!(b.0, 200, "{}", b.1);
|
||||
|
||||
let catalog = windmill_common::datatable_roles::read_role_catalog(&db).await?;
|
||||
let catalog = windmill_common::datatable_roles::read_role_catalog(
|
||||
&db,
|
||||
windmill_common::datatable_roles::DatatableRoleCluster::Instance,
|
||||
)
|
||||
.await?;
|
||||
let recorded: Vec<&str> = catalog.values().map(|r| r.name.as_str()).collect();
|
||||
for name in &names {
|
||||
assert!(
|
||||
@@ -460,7 +464,11 @@ async fn a_role_delete_that_fails_part_way_leaves_the_role_disabled(
|
||||
let body = resp.text().await?;
|
||||
assert_eq!(status, 400, "{body}");
|
||||
|
||||
let catalog = windmill_common::datatable_roles::read_role_catalog(&db).await?;
|
||||
let catalog = windmill_common::datatable_roles::read_role_catalog(
|
||||
&db,
|
||||
windmill_common::datatable_roles::DatatableRoleCluster::Instance,
|
||||
)
|
||||
.await?;
|
||||
let role = catalog
|
||||
.get(&id)
|
||||
.expect("a failed delete keeps the entry to retry");
|
||||
|
||||
@@ -176,6 +176,14 @@ pub fn global_service() -> Router {
|
||||
"/external_instance_pg/setup",
|
||||
post(setup_external_instance_pg),
|
||||
)
|
||||
.route(
|
||||
"/external_instance_pg/databases",
|
||||
get(list_external_instance_pg_databases),
|
||||
)
|
||||
.route(
|
||||
"/external_instance_pg/databases/{name}",
|
||||
post(create_external_instance_pg_database).delete(drop_external_instance_pg_database),
|
||||
)
|
||||
.route(
|
||||
"/setup_custom_instance_pg_database/{name}",
|
||||
post(setup_custom_instance_pg_database),
|
||||
@@ -885,6 +893,14 @@ pub async fn set_global_setting_internal(
|
||||
)));
|
||||
}
|
||||
|
||||
if key == EXTERNAL_INSTANCE_PG_SETTING {
|
||||
return windmill_common::external_instance_pg::write_external_instance_pg_setting(
|
||||
db,
|
||||
Some(&value),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
run_setting_pre_write_hook(db, &key, &value).await?;
|
||||
|
||||
match value {
|
||||
@@ -946,13 +962,6 @@ async fn run_setting_pre_write_hook(
|
||||
value: &serde_json::Value,
|
||||
) -> error::Result<()> {
|
||||
match key {
|
||||
EXTERNAL_INSTANCE_PG_SETTING => {
|
||||
windmill_common::external_instance_pg::check_external_instance_pg_write(
|
||||
db,
|
||||
Some(value),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
// The instance AI config is written as an untyped blob through this generic
|
||||
// endpoint, so it never passes the typed check the workspace handler applies.
|
||||
// Rates that reach a cost total unbounded would make it negative or infinite.
|
||||
@@ -1273,7 +1282,7 @@ async fn set_instance_config(
|
||||
let desired_map = desired.global_settings.to_settings_map();
|
||||
if !desired_map.is_empty() {
|
||||
let current_map = current.global_settings.to_settings_map();
|
||||
let settings_diff =
|
||||
let mut settings_diff =
|
||||
instance_config::diff_global_settings(¤t_map, &desired_map, ApplyMode::Merge);
|
||||
let ai_config_changed = settings_diff
|
||||
.upserts
|
||||
@@ -1302,16 +1311,15 @@ async fn set_instance_config(
|
||||
}
|
||||
|
||||
for (key, value) in &settings_diff.upserts {
|
||||
run_setting_pre_write_hook(&db, key, value).await?;
|
||||
}
|
||||
if settings_diff
|
||||
.deletes
|
||||
.iter()
|
||||
.any(|k| k == EXTERNAL_INSTANCE_PG_SETTING)
|
||||
{
|
||||
windmill_common::external_instance_pg::check_external_instance_pg_write(&db, None)
|
||||
.await?;
|
||||
if key != EXTERNAL_INSTANCE_PG_SETTING {
|
||||
run_setting_pre_write_hook(&db, key, value).await?;
|
||||
}
|
||||
}
|
||||
windmill_common::external_instance_pg::write_external_instance_pg_from_diff(
|
||||
&db,
|
||||
&mut settings_diff,
|
||||
)
|
||||
.await?;
|
||||
|
||||
instance_config::apply_settings_diff(&db, &settings_diff)
|
||||
.await
|
||||
@@ -1814,6 +1822,87 @@ async fn setup_external_instance_pg(
|
||||
Ok(Json(report))
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct ExternalInstancePgDatabase {
|
||||
#[serde(flatten)]
|
||||
status: windmill_common::instance_config::CustomInstanceDb,
|
||||
used_by_workspaces: Vec<String>,
|
||||
}
|
||||
|
||||
async fn list_external_instance_pg_databases(
|
||||
authed: ApiAuthed,
|
||||
Extension(db): Extension<DB>,
|
||||
) -> JsonResult<std::collections::BTreeMap<String, ExternalInstancePgDatabase>> {
|
||||
require_super_admin(&db, &authed).await?;
|
||||
let databases = windmill_common::external_instance_pg::external_instance_databases(&db).await?;
|
||||
let mut usages =
|
||||
windmill_common::external_instance_pg::external_instance_database_usages(&db).await?;
|
||||
Ok(Json(
|
||||
databases
|
||||
.into_iter()
|
||||
.map(|(name, status)| {
|
||||
let used_by_workspaces = usages.remove(&name).unwrap_or_default();
|
||||
(
|
||||
name,
|
||||
ExternalInstancePgDatabase {
|
||||
status,
|
||||
used_by_workspaces: used_by_workspaces.into_iter().collect(),
|
||||
},
|
||||
)
|
||||
})
|
||||
.collect(),
|
||||
))
|
||||
}
|
||||
|
||||
async fn create_external_instance_pg_database(
|
||||
authed: ApiAuthed,
|
||||
Extension(db): Extension<DB>,
|
||||
Path(dbname): Path<String>,
|
||||
Json(body): Json<SetupCustomInstanceDbBody>,
|
||||
) -> JsonResult<()> {
|
||||
require_super_admin(&db, &authed).await?;
|
||||
let tag = body.tag.as_deref().unwrap_or("datatable");
|
||||
windmill_common::external_instance_pg::create_external_instance_database_unchecked(
|
||||
&db, &dbname, tag,
|
||||
)
|
||||
.await?;
|
||||
windmill_audit::audit_oss::audit_log(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.create_external_instance_pg_database",
|
||||
windmill_audit::ActionKind::Create,
|
||||
"global",
|
||||
Some(&authed.email),
|
||||
Some([("dbname", dbname.as_str()), ("tag", tag)].into()),
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(()))
|
||||
}
|
||||
|
||||
async fn drop_external_instance_pg_database(
|
||||
authed: ApiAuthed,
|
||||
Extension(db): Extension<DB>,
|
||||
Path(dbname): Path<String>,
|
||||
) -> JsonResult<()> {
|
||||
require_super_admin(&db, &authed).await?;
|
||||
// A data table naming a dropped database fails on every job, far from the drop that caused it.
|
||||
windmill_common::external_instance_pg::drop_external_instance_database_unchecked(
|
||||
&db, &dbname, None,
|
||||
)
|
||||
.await?;
|
||||
windmill_audit::audit_oss::audit_log(
|
||||
&db,
|
||||
&authed,
|
||||
"settings.drop_external_instance_pg_database",
|
||||
windmill_audit::ActionKind::Delete,
|
||||
"global",
|
||||
Some(&authed.email),
|
||||
Some([("dbname", dbname.as_str())].into()),
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(()))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct SetupCustomInstanceDbBody {
|
||||
tag: Option<String>,
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
* LICENSE-AGPL for a copy of the license.
|
||||
*/
|
||||
|
||||
//! Ownership and grants on the objects of an instance data table.
|
||||
//! Ownership and grants on the objects of a data table on a cluster Windmill manages.
|
||||
//!
|
||||
//! [`datatable_permissions`](crate::datatable_permissions) decides who may connect as which role;
|
||||
//! this decides what each role may then touch. Every change is a real `GRANT`, `REVOKE`,
|
||||
@@ -33,7 +33,7 @@ use windmill_audit::audit_oss::audit_log;
|
||||
use windmill_audit::ActionKind;
|
||||
use windmill_common::datatable_roles::{
|
||||
lock_role_catalog, quote_ident, read_role_catalog, read_role_catalog_tx, DatatableRoleCatalog,
|
||||
ADMIN_DATATABLE_ROLE, CUSTOM_INSTANCE_USER,
|
||||
DatatableRoleCluster, ADMIN_DATATABLE_ROLE, CUSTOM_INSTANCE_USER,
|
||||
};
|
||||
use windmill_common::error::{pg_error_message, Error, JsonResult, Result};
|
||||
use windmill_common::workspaces::{resolve_governing_datatable, DataTable, GoverningDatatable};
|
||||
@@ -295,22 +295,22 @@ fn role_names(catalog: &DatatableRoleCatalog) -> Vec<String> {
|
||||
names
|
||||
}
|
||||
|
||||
fn ensure_instance(governing: &GoverningDatatable) -> Result<()> {
|
||||
if governing.is_instance() {
|
||||
return Ok(());
|
||||
}
|
||||
Err(Error::BadRequest(format!(
|
||||
"Data table '{}' is backed by a Postgres resource, so its access is managed on that \
|
||||
server directly. Only a data table on the Windmill instance's own database has data \
|
||||
table roles to grant to.",
|
||||
governing.name
|
||||
)))
|
||||
/// The cluster whose roles the data table's grants name.
|
||||
fn ensure_managed(governing: &GoverningDatatable) -> Result<DatatableRoleCluster> {
|
||||
governing.role_cluster().ok_or_else(|| {
|
||||
Error::BadRequest(format!(
|
||||
"Data table '{}' is backed by a Postgres resource, so its access is managed on that \
|
||||
server directly. Only a data table on a database Windmill manages has data table \
|
||||
roles to grant to.",
|
||||
governing.name
|
||||
))
|
||||
})
|
||||
}
|
||||
|
||||
/// The data table's `admin` connection, and the notices Postgres sends on it.
|
||||
///
|
||||
/// Authorization: connects as `custom_instance_user` with the instance's own credentials and checks
|
||||
/// nothing. Callers MUST have authorized the request first — a request about to be refused must
|
||||
/// Authorization: connects as `custom_instance_user` with the cluster's stored credentials and
|
||||
/// checks nothing. Callers MUST have authorized the request first — a request about to be refused must
|
||||
/// not get as far as this connection.
|
||||
async fn connect_as_admin_unchecked(
|
||||
db: &DB,
|
||||
@@ -320,21 +320,33 @@ async fn connect_as_admin_unchecked(
|
||||
mpsc::UnboundedReceiver<DbError>,
|
||||
String,
|
||||
)> {
|
||||
ensure_instance(governing)?;
|
||||
let cluster = ensure_managed(governing)?;
|
||||
// Built from the authorized entry, never by resolving the settings again: a save in between
|
||||
// could point the entry at a resource on another server and back, and this connection would
|
||||
// then alter a database the later checks of the entry never see.
|
||||
let mut pg = PgDatabase::parse_uri(&windmill_common::get_database_url().await?.as_str().await)?;
|
||||
pg.dbname = governing
|
||||
let dbname = governing
|
||||
.datatable
|
||||
.database
|
||||
.as_ref()
|
||||
.expect("a governing entry owns a database")
|
||||
.resource_path
|
||||
.clone();
|
||||
pg.user = Some(CUSTOM_INSTANCE_USER.to_string());
|
||||
pg.password = Some(windmill_common::utils::get_custom_pg_instance_password(db).await?);
|
||||
let dbname = pg.dbname.clone();
|
||||
let pg = match cluster {
|
||||
DatatableRoleCluster::Instance => {
|
||||
let mut pg =
|
||||
PgDatabase::parse_uri(&windmill_common::get_database_url().await?.as_str().await)?;
|
||||
pg.dbname = dbname.clone();
|
||||
pg.user = Some(CUSTOM_INSTANCE_USER.to_string());
|
||||
pg.password = Some(windmill_common::utils::get_custom_pg_instance_password(db).await?);
|
||||
pg
|
||||
}
|
||||
DatatableRoleCluster::ExternalInstance => {
|
||||
windmill_common::external_instance_pg::external_instance_connection_unchecked(
|
||||
db, &dbname, false,
|
||||
)
|
||||
.await?
|
||||
}
|
||||
};
|
||||
let (client, mut connection) = pg.connect(Some(db)).await?;
|
||||
// Unbounded: the driver must never wait on the receiver, which only drains once the statement
|
||||
// the driver is carrying has completed.
|
||||
@@ -1020,12 +1032,12 @@ async fn get_datatable_acl(
|
||||
let target: AclTarget = query.try_into()?;
|
||||
let governing = resolve_governing_datatable(&db, &w_id, &datatable_name).await?;
|
||||
ensure_reaches_governing_datatable(&db, &w_id, &datatable_name, &governing, &authed).await?;
|
||||
ensure_instance(&governing)?;
|
||||
let cluster = ensure_managed(&governing)?;
|
||||
let editable = ensure_governs_datatable(&db, &authed, &w_id, &governing)
|
||||
.await
|
||||
.is_ok();
|
||||
let roles = if editable {
|
||||
role_names(&read_role_catalog(&db).await?)
|
||||
role_names(&read_role_catalog(&db, cluster).await?)
|
||||
} else {
|
||||
vec![]
|
||||
};
|
||||
@@ -1310,7 +1322,7 @@ async fn authorize_acl_change(
|
||||
) -> Result<GoverningDatatable> {
|
||||
let governing = resolve_governing_datatable(db, w_id, datatable_name).await?;
|
||||
ensure_governs_datatable(db, authed, w_id, &governing).await?;
|
||||
ensure_instance(&governing)?;
|
||||
ensure_managed(&governing)?;
|
||||
Ok(governing)
|
||||
}
|
||||
|
||||
@@ -1320,8 +1332,14 @@ static APPLY_SLOT: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(1)
|
||||
/// provisioned before data table roles gave `custom_instance_user` none. Adds that option to its
|
||||
/// database and `public` privileges, and nothing else: default privileges are left alone, since a
|
||||
/// schema's change of owner is planned against them. Best-effort, as a grant it fails to enable is
|
||||
/// refused when it runs.
|
||||
async fn ensure_grant_options(client: &tokio_postgres::Client, db: &DB, dbname: &str) {
|
||||
/// refused when it runs. An external instance database was created with the options, so one
|
||||
/// missing there is someone's deliberate revoke and is left alone.
|
||||
async fn ensure_grant_options(
|
||||
client: &tokio_postgres::Client,
|
||||
db: &DB,
|
||||
cluster: DatatableRoleCluster,
|
||||
dbname: &str,
|
||||
) {
|
||||
let held = client
|
||||
.query_one(
|
||||
"SELECT has_database_privilege(current_database(), 'CONNECT WITH GRANT OPTION')
|
||||
@@ -1333,7 +1351,7 @@ async fn ensure_grant_options(client: &tokio_postgres::Client, db: &DB, dbname:
|
||||
)
|
||||
.await
|
||||
.is_ok_and(|row| row.get::<_, bool>(0));
|
||||
if held {
|
||||
if held || cluster != DatatableRoleCluster::Instance {
|
||||
return;
|
||||
}
|
||||
if let Err(e) = grant_options_as_server(db, dbname).await {
|
||||
@@ -1559,7 +1577,8 @@ async fn plan_datatable_acl(
|
||||
) -> JsonResult<AclPlan> {
|
||||
crate::datatable_acl_oss::ensure_datatable_acl_available()?;
|
||||
let governing = authorize_acl_change(&db, &authed, &w_id, &datatable_name).await?;
|
||||
let catalog = read_role_catalog(&db).await?;
|
||||
let cluster = ensure_managed(&governing)?;
|
||||
let catalog = read_role_catalog(&db, cluster).await?;
|
||||
let (client, _notices, dbname) = connect_as_admin_unchecked(&db, &governing).await?;
|
||||
Ok(Json(
|
||||
build_plan(&client, &dbname, &catalog, &req.target, &req.change).await?,
|
||||
@@ -1583,6 +1602,7 @@ async fn apply_datatable_acl(
|
||||
// connection could wait forever on a pool that concurrent applies, queued on the same locks,
|
||||
// have exhausted.
|
||||
let governing = authorize_acl_change(&db, &authed, &w_id, &datatable_name).await?;
|
||||
let cluster = ensure_managed(&governing)?;
|
||||
// Applies queue on an instance-wide lock while each holds a direct connection to the instance's
|
||||
// Postgres; unbounded, the queue alone could exhaust its connection limit. One at a time per
|
||||
// server, and the ones waiting hold no connection at all.
|
||||
@@ -1591,7 +1611,7 @@ async fn apply_datatable_acl(
|
||||
.await
|
||||
.map_err(|e| Error::internal_err(format!("ACL apply slot closed: {e}")))?;
|
||||
let (mut client, mut notices, dbname) = connect_as_admin_unchecked(&db, &governing).await?;
|
||||
ensure_grant_options(&client, &db, &dbname).await;
|
||||
ensure_grant_options(&client, &db, cluster, &dbname).await;
|
||||
|
||||
// Held until the change is committed: a role renamed or dropped meanwhile would change what
|
||||
// the plan names, and a settings save could move the entry onto another database. Taken in the
|
||||
@@ -1606,7 +1626,7 @@ async fn apply_datatable_acl(
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?
|
||||
.flatten();
|
||||
let catalog = read_role_catalog_tx(&mut tx).await?;
|
||||
let catalog = read_role_catalog_tx(&mut tx, cluster).await?;
|
||||
|
||||
let plan = build_plan(&client, &dbname, &catalog, &req.target, &req.change).await?;
|
||||
if !entry_unchanged(&governing, entry_now) || &plan.statements != confirmed {
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
//! to keep that file focused on core workspace configuration.
|
||||
|
||||
use crate::workspaces::{
|
||||
is_instance_datatable, pg_dump_database, strip_unreplayable_dump_lines, ItemComparison,
|
||||
managed_datatable_kind, pg_dump_database, strip_unreplayable_dump_lines, ItemComparison,
|
||||
PgDumpOptions,
|
||||
};
|
||||
|
||||
@@ -1556,7 +1556,9 @@ async fn generate_initial_datatable_migration(
|
||||
// without what a replay elsewhere cannot run: the replaying user owns none of this
|
||||
// database's objects, and the grants Windmill plants in an instance database (`ALTER
|
||||
// DEFAULT PRIVILEGES FOR ROLE ...`) fail even replaying onto the same server.
|
||||
let no_acl = is_instance_datatable(&db, &w_id, &datatable_name).await?;
|
||||
let no_acl = managed_datatable_kind(&db, &w_id, &datatable_name)
|
||||
.await?
|
||||
.is_some();
|
||||
let dump_file = pg_dump_database(
|
||||
&pg_db,
|
||||
PgDumpOptions {
|
||||
|
||||
@@ -2265,7 +2265,8 @@ struct DataTableTables {
|
||||
schemas: TableListMap,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
error: Option<String>,
|
||||
/// On the instance database: the only kind that can be under roles or have its access edited.
|
||||
/// On a database Windmill manages, on its own cluster or the external one: the only kinds that
|
||||
/// can be under roles or have their access edited.
|
||||
instance: bool,
|
||||
permissioned: bool,
|
||||
/// The roles this caller may connect as, by name; empty when not under roles.
|
||||
@@ -2522,7 +2523,7 @@ async fn list_one_datatable_tables(
|
||||
};
|
||||
let result: Result<()> = async {
|
||||
let governing = resolve_governing_datatable(db, w_id, &entry.datatable_name).await?;
|
||||
entry.instance = governing.is_instance();
|
||||
entry.instance = governing.role_cluster().is_some();
|
||||
let usable =
|
||||
crate::datatable_permissions_oss::usable_datatable_roles(db, authed, w_id, &governing)
|
||||
.await?;
|
||||
@@ -3080,23 +3081,32 @@ pub(crate) async fn resolve_pg_source_checked(
|
||||
.map_err(|e| Error::internal_err(format!("Failed to parse database credentials: {}", e)))
|
||||
}
|
||||
|
||||
/// Whether the data table `name` is backed by the Windmill instance's own PostgreSQL
|
||||
/// rather than a user resource.
|
||||
pub(crate) async fn is_instance_datatable(db: &DB, w_id: &str, name: &str) -> Result<bool> {
|
||||
/// The kind of the database backing the data table `name` when Windmill manages it (on its own
|
||||
/// cluster or the external one), `None` when it is a user resource.
|
||||
pub(crate) async fn managed_datatable_kind(
|
||||
db: &DB,
|
||||
w_id: &str,
|
||||
name: &str,
|
||||
) -> Result<Option<DataTableCatalogResourceType>> {
|
||||
// Resolved rather than read: a pointer entry owns no database of its own, so only the entry it
|
||||
// lands on can answer. A name that resolves to nothing keeps the historical `false`.
|
||||
// lands on can answer. A name that resolves to nothing keeps the historical `None`.
|
||||
Ok(resolve_governing_datatable(db, w_id, name)
|
||||
.await
|
||||
.ok()
|
||||
.and_then(|g| g.datatable.database)
|
||||
.is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance))
|
||||
.map(|d| d.resource_type)
|
||||
.filter(|kind| kind.is_windmill_managed()))
|
||||
}
|
||||
|
||||
/// Same, for the `datatable://<name>` / `$res:<path>` form the import endpoints take.
|
||||
async fn is_instance_datatable_source(db: &DB, w_id: &str, source: &str) -> Result<bool> {
|
||||
async fn managed_datatable_source_kind(
|
||||
db: &DB,
|
||||
w_id: &str,
|
||||
source: &str,
|
||||
) -> Result<Option<DataTableCatalogResourceType>> {
|
||||
match source.strip_prefix("datatable://") {
|
||||
Some(name) => is_instance_datatable(db, w_id, name).await,
|
||||
None => Ok(false),
|
||||
Some(name) => managed_datatable_kind(db, w_id, name).await,
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3195,10 +3205,7 @@ pub(crate) async fn pg_dump_database(
|
||||
if let Some(ref password) = pg_db.password {
|
||||
cmd.env("PGPASSWORD", password);
|
||||
}
|
||||
|
||||
if let Some(ref sslmode) = pg_db.sslmode {
|
||||
cmd.env("PGSSLMODE", sslmode);
|
||||
}
|
||||
let _root_cert = apply_pg_tls_env(&mut cmd, pg_db)?;
|
||||
|
||||
let output = cmd
|
||||
.output()
|
||||
@@ -3316,7 +3323,7 @@ async fn comment_out_unsupported_settings(
|
||||
|
||||
/// A psql invocation against `pg_db`, carrying the connection settings the CLI reads
|
||||
/// from the environment.
|
||||
fn psql_command(pg_db: &PgDatabase) -> tokio::process::Command {
|
||||
fn psql_command(pg_db: &PgDatabase) -> Result<(tokio::process::Command, Option<DumpFile>)> {
|
||||
let mut cmd = tokio::process::Command::new("psql");
|
||||
cmd.arg("--host")
|
||||
.arg(&pg_db.host)
|
||||
@@ -3333,10 +3340,88 @@ fn psql_command(pg_db: &PgDatabase) -> tokio::process::Command {
|
||||
if let Some(ref password) = pg_db.password {
|
||||
cmd.env("PGPASSWORD", password);
|
||||
}
|
||||
let root_cert = apply_pg_tls_env(&mut cmd, pg_db)?;
|
||||
Ok((cmd, root_cert))
|
||||
}
|
||||
|
||||
/// Give libpq the TLS settings `PgDatabase::connect` applies. The returned file holds the root
|
||||
/// certificate `PGSSLROOTCERT` names, so it must outlive the command.
|
||||
fn apply_pg_tls_env(
|
||||
cmd: &mut tokio::process::Command,
|
||||
pg_db: &PgDatabase,
|
||||
) -> Result<Option<DumpFile>> {
|
||||
if let Some(ref sslmode) = pg_db.sslmode {
|
||||
cmd.env("PGSSLMODE", sslmode);
|
||||
}
|
||||
cmd
|
||||
if let Some(pem) = pg_db
|
||||
.root_certificate_pem
|
||||
.as_deref()
|
||||
.filter(|p| !p.is_empty())
|
||||
{
|
||||
let file = DumpFile::new()?;
|
||||
std::fs::write(&file.path, pem)
|
||||
.map_err(|e| Error::internal_err(format!("Failed to write root certificate: {e}")))?;
|
||||
cmd.env("PGSSLROOTCERT", &file.path);
|
||||
return Ok(Some(file));
|
||||
}
|
||||
// Only a connection that asked to be verified against the system trust store. Without a file,
|
||||
// libpq's own default would look for `~/.postgresql/root.crt` and refuse a verify-* mode. libpq
|
||||
// takes the special `system` value with verify-full only, so verify-ca needs the bundle itself.
|
||||
if pg_db.accept_invalid_certs == Some(false) {
|
||||
match pg_db.sslmode.as_deref() {
|
||||
Some("verify-full") => {
|
||||
cmd.env("PGSSLROOTCERT", "system");
|
||||
}
|
||||
Some("verify-ca") => {
|
||||
if let Some(bundle) = windmill_common::system_ca_bundle() {
|
||||
cmd.env("PGSSLROOTCERT", bundle);
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
|
||||
#[cfg(test)]
|
||||
mod pg_tls_env_tests {
|
||||
use super::apply_pg_tls_env;
|
||||
use windmill_common::PgDatabase;
|
||||
|
||||
fn root_cert_env(sslmode: &str) -> Option<std::ffi::OsString> {
|
||||
let pg_db = PgDatabase {
|
||||
host: "db".to_string(),
|
||||
user: None,
|
||||
password: None,
|
||||
port: None,
|
||||
sslmode: Some(sslmode.to_string()),
|
||||
dbname: "d".to_string(),
|
||||
root_certificate_pem: None,
|
||||
accept_invalid_certs: Some(false),
|
||||
use_iam_auth: None,
|
||||
region: None,
|
||||
};
|
||||
let mut cmd = tokio::process::Command::new("psql");
|
||||
apply_pg_tls_env(&mut cmd, &pg_db).unwrap();
|
||||
cmd.as_std()
|
||||
.get_envs()
|
||||
.find(|(k, _)| *k == "PGSSLROOTCERT")
|
||||
.and_then(|(_, v)| v.map(|v| v.to_os_string()))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn system_roots_only_through_verify_full() {
|
||||
assert_eq!(
|
||||
root_cert_env("verify-full").as_deref(),
|
||||
Some("system".as_ref())
|
||||
);
|
||||
// libpq refuses `sslrootcert=system` with verify-ca, which would fail every dump and restore.
|
||||
assert_ne!(
|
||||
root_cert_env("verify-ca").as_deref(),
|
||||
Some("system".as_ref())
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// GUC names the server backing `pg_db` knows about.
|
||||
@@ -3346,7 +3431,8 @@ fn psql_command(pg_db: &PgDatabase) -> tokio::process::Command {
|
||||
/// and an unset mode, where `PgDatabase::connect` would hand a TLS-only server a
|
||||
/// plaintext socket and fail before the import ever starts.
|
||||
async fn server_setting_names(pg_db: &PgDatabase) -> Result<HashSet<String>> {
|
||||
let output = psql_command(pg_db)
|
||||
let (mut cmd, _root_cert) = psql_command(pg_db)?;
|
||||
let output = cmd
|
||||
.arg("--tuples-only")
|
||||
.arg("--no-align")
|
||||
.arg("--command")
|
||||
@@ -3380,7 +3466,8 @@ async fn pg_import_dump(target_db: &PgDatabase, dump_file: &DumpFile) -> Result<
|
||||
let supported_settings = server_setting_names(target_db).await?;
|
||||
comment_out_unsupported_settings(dump_file, &supported_settings).await?;
|
||||
|
||||
let output = psql_command(target_db)
|
||||
let (mut cmd, _root_cert) = psql_command(target_db)?;
|
||||
let output = cmd
|
||||
.arg("--set")
|
||||
.arg("ON_ERROR_STOP=1")
|
||||
.arg("--single-transaction")
|
||||
@@ -3438,7 +3525,15 @@ async fn create_pg_database(
|
||||
}
|
||||
}
|
||||
|
||||
if is_instance_datatable_source(&db, &w_id, &req.source).await? {
|
||||
let source_kind = managed_datatable_source_kind(&db, &w_id, &req.source).await?;
|
||||
if source_kind == Some(DataTableCatalogResourceType::ExternalInstance) {
|
||||
windmill_common::external_instance_pg::create_external_instance_database_unchecked(
|
||||
&db,
|
||||
&req.target_dbname,
|
||||
"datatable",
|
||||
)
|
||||
.await?;
|
||||
} else if source_kind == Some(DataTableCatalogResourceType::Instance) {
|
||||
windmill_common::create_custom_instance_database(&db, &req.target_dbname, "datatable")
|
||||
.await?;
|
||||
} else {
|
||||
@@ -3541,12 +3636,12 @@ async fn ensure_datatable_is_clonable(
|
||||
}
|
||||
// The copy has to name a database of its own. A resource-backed entry reached through a
|
||||
// pointer names one this workspace does not own, so there is nothing here to repoint.
|
||||
let is_instance = governing
|
||||
let is_managed = governing
|
||||
.datatable
|
||||
.database
|
||||
.as_ref()
|
||||
.is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance);
|
||||
if governing.workspace_id != w_id && !is_instance {
|
||||
.is_some_and(|d| d.resource_type.is_windmill_managed());
|
||||
if governing.workspace_id != w_id && !is_managed {
|
||||
return Err(Error::BadRequest(format!(
|
||||
"Data table '{name}' points at a resource-backed data table in another workspace \
|
||||
and cannot be copied; fork it from the workspace that owns it."
|
||||
@@ -3603,8 +3698,12 @@ async fn import_pg_database(
|
||||
// what it creates it owns. Grants do, except around an instance data table — Windmill
|
||||
// plants `custom_instance_user` grants in one, which nothing else can replay. Elsewhere
|
||||
// the ACLs are user intent (`REVOKE ... FROM PUBLIC`) and dropping them widens access.
|
||||
let no_acl = is_instance_datatable_source(&db, &w_id, &req.target).await?
|
||||
|| is_instance_datatable_source(&db, &w_id, &req.source).await?;
|
||||
let no_acl = managed_datatable_source_kind(&db, &w_id, &req.target)
|
||||
.await?
|
||||
.is_some()
|
||||
|| managed_datatable_source_kind(&db, &w_id, &req.source)
|
||||
.await?
|
||||
.is_some();
|
||||
|
||||
let dump_file = pg_dump_database(
|
||||
&source_pg,
|
||||
@@ -3717,22 +3816,37 @@ async fn edit_ducklake_config(
|
||||
let old_ducklakes: HashMap<String, Ducklake> =
|
||||
serde_json::from_value(old_ducklakes).unwrap_or_default();
|
||||
|
||||
// Check that non-superadmins are not abusing Instance databases
|
||||
if !is_superadmin {
|
||||
for (name, dl) in new_config.settings.ducklakes.iter() {
|
||||
if dl.catalog.resource_type == DucklakeCatalogResourceType::Instance {
|
||||
let old_dl = old_ducklakes.get(name);
|
||||
if old_dl.is_none()
|
||||
|| old_dl.unwrap().catalog.resource_type
|
||||
!= DucklakeCatalogResourceType::Instance
|
||||
|| old_dl.unwrap().catalog.resource_path != dl.catalog.resource_path
|
||||
{
|
||||
return Err(Error::BadRequest(
|
||||
"Only superadmins can create or modify ducklakes with Instance databases"
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
}
|
||||
// Check that non-superadmins are not abusing Instance databases. An unchanged catalog is left
|
||||
// alone either way, so a downgraded instance can still save lakes that already name an
|
||||
// external instance database.
|
||||
for (name, dl) in new_config.settings.ducklakes.iter() {
|
||||
let kind = &dl.catalog.resource_type;
|
||||
if !matches!(
|
||||
kind,
|
||||
DucklakeCatalogResourceType::Instance | DucklakeCatalogResourceType::ExternalInstance
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
let unchanged = old_ducklakes.get(name).is_some_and(|old| {
|
||||
&old.catalog.resource_type == kind
|
||||
&& old.catalog.resource_path == dl.catalog.resource_path
|
||||
});
|
||||
if unchanged {
|
||||
continue;
|
||||
}
|
||||
if *kind == DucklakeCatalogResourceType::ExternalInstance {
|
||||
windmill_common::external_instance_pg::ensure_external_instance_available()?;
|
||||
windmill_common::external_instance_pg::ensure_external_instance_database_registered(
|
||||
&mut tx,
|
||||
&dl.catalog.resource_path,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
if !is_superadmin {
|
||||
return Err(Error::BadRequest(
|
||||
"Only superadmins can create or modify ducklakes with Instance databases"
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3904,6 +4018,7 @@ async fn edit_datatable_config(
|
||||
// so these line up with the `datatable_configured` adoption counts.
|
||||
created_substrates.push(match dt.database.as_ref().map(|d| d.resource_type) {
|
||||
Some(DataTableCatalogResourceType::Instance) => "instance",
|
||||
Some(DataTableCatalogResourceType::ExternalInstance) => "external_instance",
|
||||
Some(DataTableCatalogResourceType::Postgresql) => "postgresql",
|
||||
None => "reference",
|
||||
});
|
||||
@@ -3923,18 +4038,20 @@ async fn edit_datatable_config(
|
||||
None => None,
|
||||
};
|
||||
// Carrying the block onto a resource-backed entry would produce a data table the chokepoint
|
||||
// refuses on every job — a save that succeeds and breaks everything afterwards. Refuse it
|
||||
// instead: turning roles off first is one step, and it keeps discarding an access decision
|
||||
// something somebody chose rather than a side effect of moving a database.
|
||||
// refuses on every job — a save that succeeds and breaks everything afterwards — and onto
|
||||
// the other managed cluster, one whose role ids name nothing in that cluster's catalog.
|
||||
// Refuse it instead: turning roles off first is one step, and it keeps discarding an access
|
||||
// decision something somebody chose rather than a side effect of moving a database.
|
||||
let old_kind = old.and_then(|old| old.database.as_ref()).map(|d| d.resource_type);
|
||||
if dt.permissions.is_some()
|
||||
&& dt
|
||||
.database
|
||||
.as_ref()
|
||||
.is_some_and(|d| d.resource_type != DataTableCatalogResourceType::Instance)
|
||||
.is_some_and(|d| Some(d.resource_type) != old_kind)
|
||||
{
|
||||
return Err(Error::BadRequest(format!(
|
||||
"Data table '{name}' is under roles, which only a data table on the instance \
|
||||
database can be. Turn its roles off before moving it to a PostgreSQL resource."
|
||||
"Data table '{name}' is under roles, which belong to the cluster its database is \
|
||||
on. Turn its roles off before moving it to another kind of database."
|
||||
)));
|
||||
}
|
||||
// A pointer names no database of its own, so the form's empty `database` is correct there.
|
||||
@@ -3959,26 +4076,39 @@ async fn edit_datatable_config(
|
||||
// Check that non-superadmins are not abusing Instance databases, which reach a database this
|
||||
// workspace does not own. Pointing an entry at another workspace's data table is not checked
|
||||
// here because it cannot be requested at all: `reference` is overwritten from the stored entry
|
||||
// above, for every caller.
|
||||
if !is_superadmin {
|
||||
for (name, dt) in new_config.settings.datatables.iter() {
|
||||
let old_dt = old_datatables.get(name);
|
||||
if dt
|
||||
.database
|
||||
.as_ref()
|
||||
.is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance)
|
||||
{
|
||||
let unchanged = old_dt.and_then(|o| o.database.as_ref()).is_some_and(|o| {
|
||||
o.resource_type == DataTableCatalogResourceType::Instance
|
||||
&& Some(&o.resource_path) == dt.database.as_ref().map(|d| &d.resource_path)
|
||||
});
|
||||
if !unchanged {
|
||||
return Err(Error::BadRequest(
|
||||
"Only superadmins can create or modify data tables with Instance databases"
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
}
|
||||
// above, for every caller. An unchanged entry is left alone either way, so a downgraded
|
||||
// instance can still save settings that already name an external instance database.
|
||||
for (name, dt) in new_config.settings.datatables.iter() {
|
||||
let Some(database) = dt
|
||||
.database
|
||||
.as_ref()
|
||||
.filter(|d| d.resource_type.is_windmill_managed())
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let unchanged = old_datatables
|
||||
.get(name)
|
||||
.and_then(|o| o.database.as_ref())
|
||||
.is_some_and(|o| {
|
||||
o.resource_type == database.resource_type
|
||||
&& o.resource_path == database.resource_path
|
||||
});
|
||||
if unchanged {
|
||||
continue;
|
||||
}
|
||||
if database.resource_type == DataTableCatalogResourceType::ExternalInstance {
|
||||
windmill_common::external_instance_pg::ensure_external_instance_available()?;
|
||||
windmill_common::external_instance_pg::ensure_external_instance_database_registered(
|
||||
&mut tx,
|
||||
&database.resource_path,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
if !is_superadmin {
|
||||
return Err(Error::BadRequest(
|
||||
"Only superadmins can create or modify data tables with Instance databases"
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3998,7 +4128,7 @@ async fn edit_datatable_config(
|
||||
// entry through a declared rename alone, and a settings sync never declares one, so an entry
|
||||
// without roles that newly points at such a database — a name added, or an existing one
|
||||
// repointed — would answer everyone there as `admin`. That holds whichever workspace governs it.
|
||||
let newly_pointed: Vec<(&String, &str)> = new_config
|
||||
let newly_pointed: Vec<(&String, DataTableCatalogResourceType, &str)> = new_config
|
||||
.settings
|
||||
.datatables
|
||||
.iter()
|
||||
@@ -4007,7 +4137,7 @@ async fn edit_datatable_config(
|
||||
let db = dt
|
||||
.database
|
||||
.as_ref()
|
||||
.filter(|d| d.resource_type == DataTableCatalogResourceType::Instance)?;
|
||||
.filter(|d| d.resource_type.is_windmill_managed())?;
|
||||
let lookup = rename_src
|
||||
.get(name.as_str())
|
||||
.copied()
|
||||
@@ -4019,38 +4149,44 @@ async fn edit_datatable_config(
|
||||
old_db.resource_type != db.resource_type
|
||||
|| old_db.resource_path != db.resource_path
|
||||
});
|
||||
repointed.then_some((name, db.resource_path.as_str()))
|
||||
repointed.then_some((name, db.resource_type, db.resource_path.as_str()))
|
||||
})
|
||||
.collect();
|
||||
// Another workspace turning roles on for the same database holds only its own settings row, so
|
||||
// without this the scan below could read past its uncommitted write.
|
||||
windmill_common::datatable_roles::lock_instance_databases_governance(
|
||||
&mut *tx,
|
||||
newly_pointed.iter().map(|(_, dbname)| *dbname),
|
||||
newly_pointed.iter().map(|(_, _, dbname)| *dbname),
|
||||
)
|
||||
.await?;
|
||||
let governed_elsewhere: Vec<String> = if newly_pointed.is_empty() {
|
||||
let governed_elsewhere: Vec<(String, String)> = if newly_pointed.is_empty() {
|
||||
vec![]
|
||||
} else {
|
||||
sqlx::query_scalar(
|
||||
"SELECT DISTINCT dt.value->'database'->>'resource_path' FROM workspace_settings ws
|
||||
sqlx::query_as(
|
||||
"SELECT DISTINCT dt.value->'database'->>'resource_type',
|
||||
dt.value->'database'->>'resource_path'
|
||||
FROM workspace_settings ws
|
||||
CROSS JOIN LATERAL jsonb_each(COALESCE(ws.datatable->'datatables', '{}'::jsonb)) dt
|
||||
WHERE ws.workspace_id <> $1 AND dt.value ? 'permissions'
|
||||
AND dt.value->'database'->>'resource_type' = 'instance'",
|
||||
AND dt.value->'database'->>'resource_type' IN ('instance', 'external_instance')",
|
||||
)
|
||||
.bind(&w_id)
|
||||
.fetch_all(&mut *tx)
|
||||
.await?
|
||||
};
|
||||
for (name, dbname) in newly_pointed {
|
||||
for (name, kind, dbname) in newly_pointed {
|
||||
let governed_here = old_datatables.values().any(|old| {
|
||||
old.permissions.is_some()
|
||||
&& old.database.as_ref().is_some_and(|d| {
|
||||
d.resource_type == DataTableCatalogResourceType::Instance
|
||||
&& d.resource_path == dbname
|
||||
})
|
||||
&& old
|
||||
.database
|
||||
.as_ref()
|
||||
.is_some_and(|d| d.resource_type == kind && d.resource_path == dbname)
|
||||
});
|
||||
if governed_here || governed_elsewhere.iter().any(|g| g == dbname) {
|
||||
if governed_here
|
||||
|| governed_elsewhere
|
||||
.iter()
|
||||
.any(|(k, p)| k == kind.as_ref() && p == dbname)
|
||||
{
|
||||
return Err(Error::BadRequest(format!(
|
||||
"Data table '{name}' would point at database '{dbname}', which a data table under \
|
||||
roles uses, without carrying those roles: everyone reaching '{name}' would connect \
|
||||
@@ -8064,13 +8200,14 @@ async fn point_kept_datatables_at_parent(
|
||||
if dt.reference.is_some() {
|
||||
continue;
|
||||
}
|
||||
// Only instance databases. A resource-backed data table names a resource, and the settings
|
||||
// clone gave the fork its own copy of that resource in its own workspace — pointing at the
|
||||
// parent's entry would silently move the fork onto the parent's resource instead.
|
||||
// Only instance databases, on either cluster. A resource-backed data table names a
|
||||
// resource, and the settings clone gave the fork its own copy of that resource in its own
|
||||
// workspace — pointing at the parent's entry would silently move the fork onto the
|
||||
// parent's resource instead.
|
||||
if dt
|
||||
.database
|
||||
.as_ref()
|
||||
.is_none_or(|d| d.resource_type != DataTableCatalogResourceType::Instance)
|
||||
.is_none_or(|d| !d.resource_type.is_windmill_managed())
|
||||
{
|
||||
continue;
|
||||
}
|
||||
@@ -8200,11 +8337,19 @@ async fn apply_forked_datatable(
|
||||
})?,
|
||||
};
|
||||
|
||||
if database.resource_type == DataTableCatalogResourceType::Instance {
|
||||
if database.resource_type == DataTableCatalogResourceType::ExternalInstance {
|
||||
windmill_common::external_instance_pg::ensure_external_instance_database_registered(
|
||||
tx,
|
||||
&fdt.new_dbname,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
if database.resource_type.is_windmill_managed() {
|
||||
// The whole `database` object, not just its `resource_path`: a pointer entry has none to
|
||||
// patch. `reference` goes with it — exactly one of the two may be set.
|
||||
// patch. `reference` goes with it — exactly one of the two may be set. The copy was created
|
||||
// on the same cluster as its source, so it keeps the source's kind.
|
||||
let new_database = serde_json::json!({
|
||||
"resource_type": "instance",
|
||||
"resource_type": database.resource_type,
|
||||
"resource_path": &fdt.new_dbname,
|
||||
});
|
||||
sqlx::query!(
|
||||
|
||||
@@ -1409,9 +1409,7 @@ pub async fn drop_forked_datatable_databases(
|
||||
_ => continue,
|
||||
};
|
||||
|
||||
if database.resource_type
|
||||
== windmill_common::workspaces::DataTableCatalogResourceType::Instance
|
||||
{
|
||||
if database.resource_type.is_windmill_managed() {
|
||||
let db_to_drop = &database.resource_path;
|
||||
if !db_to_drop.starts_with("wm_fork_") {
|
||||
errors.push(format!(
|
||||
@@ -1420,7 +1418,20 @@ pub async fn drop_forked_datatable_databases(
|
||||
));
|
||||
continue;
|
||||
}
|
||||
if let Err(e) = windmill_common::drop_custom_instance_database(&db, db_to_drop).await {
|
||||
let dropped = if database.resource_type
|
||||
== windmill_common::workspaces::DataTableCatalogResourceType::ExternalInstance
|
||||
{
|
||||
// Its own entry still names the copy; another workspace's never should.
|
||||
windmill_common::external_instance_pg::drop_external_instance_database_unchecked(
|
||||
&db,
|
||||
db_to_drop,
|
||||
Some(&w_id),
|
||||
)
|
||||
.await
|
||||
} else {
|
||||
windmill_common::drop_custom_instance_database(&db, db_to_drop).await
|
||||
};
|
||||
if let Err(e) = dropped {
|
||||
errors.push(format!(
|
||||
"Could not drop instance database '{}' for datatable://{}: {}",
|
||||
db_to_drop, dt_name, e
|
||||
@@ -1787,7 +1798,17 @@ async fn resolve_fork_catalog_pg(
|
||||
"ducklake://{ducklake_name}: malformed registry catalog identity `{catalog}`"
|
||||
))
|
||||
})?;
|
||||
let catalog_resource = if resource_type == "instance" {
|
||||
let catalog_resource = if resource_type == "external_instance" {
|
||||
serde_json::to_value(
|
||||
windmill_common::external_instance_pg::external_instance_connection_unchecked(
|
||||
db,
|
||||
resource_path,
|
||||
false,
|
||||
)
|
||||
.await?,
|
||||
)
|
||||
.map_err(|e| Error::internal_err(format!("serializing pg creds: {e}")))?
|
||||
} else if resource_type == "instance" {
|
||||
let mut pg_creds = windmill_common::PgDatabase::parse_uri(
|
||||
&windmill_common::get_database_url().await?.as_str().await,
|
||||
)?;
|
||||
|
||||
@@ -1609,6 +1609,67 @@ paths:
|
||||
schema:
|
||||
$ref: "#/components/schemas/ExternalInstancePgSetupReport"
|
||||
|
||||
/settings/external_instance_pg/databases:
|
||||
get:
|
||||
summary: Lists the databases Windmill created on the external instance cluster, with the workspaces whose data tables, Ducklake catalogs or pending fork cleanups use each
|
||||
operationId: listExternalInstancePgDatabases
|
||||
tags:
|
||||
- setting
|
||||
responses:
|
||||
"200":
|
||||
description: databases by name
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
additionalProperties:
|
||||
$ref: "#/components/schemas/CustomInstanceDb"
|
||||
|
||||
/settings/external_instance_pg/databases/{name}:
|
||||
post:
|
||||
summary: Creates a database on the external instance cluster (enterprise edition only)
|
||||
operationId: createExternalInstancePgDatabase
|
||||
tags:
|
||||
- setting
|
||||
parameters:
|
||||
- name: name
|
||||
in: path
|
||||
required: true
|
||||
schema:
|
||||
type: string
|
||||
requestBody:
|
||||
required: true
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
tag:
|
||||
$ref: "#/components/schemas/CustomInstanceDbTag"
|
||||
responses:
|
||||
"200":
|
||||
description: database created
|
||||
content:
|
||||
application/json:
|
||||
schema: {}
|
||||
delete:
|
||||
summary: Drops a database Windmill created on the external instance cluster, refused while a data table, Ducklake catalog or pending fork cleanup uses it
|
||||
operationId: dropExternalInstancePgDatabase
|
||||
tags:
|
||||
- setting
|
||||
parameters:
|
||||
- name: name
|
||||
in: path
|
||||
required: true
|
||||
schema:
|
||||
type: string
|
||||
responses:
|
||||
"200":
|
||||
description: database dropped
|
||||
content:
|
||||
application/json:
|
||||
schema: {}
|
||||
|
||||
/settings/list_custom_instance_pg_databases:
|
||||
post:
|
||||
summary: Returns the set-up statuses of custom instance pg databases
|
||||
@@ -1627,13 +1688,19 @@ paths:
|
||||
|
||||
/settings/datatable_roles:
|
||||
get:
|
||||
summary: list the instance's data table roles
|
||||
summary: list the data table roles of one Windmill-managed Postgres cluster
|
||||
operationId: listInstanceDatatableRoles
|
||||
tags:
|
||||
- setting
|
||||
parameters:
|
||||
- in: query
|
||||
name: cluster
|
||||
required: false
|
||||
schema:
|
||||
$ref: "#/components/schemas/DatatableRoleCluster"
|
||||
responses:
|
||||
"200":
|
||||
description: the instance role catalog
|
||||
description: the cluster's role catalog
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
@@ -1641,7 +1708,7 @@ paths:
|
||||
items:
|
||||
$ref: "#/components/schemas/InstanceDatatableRole"
|
||||
post:
|
||||
summary: create a data table role on the instance's Postgres cluster
|
||||
summary: create a data table role on a Windmill-managed Postgres cluster
|
||||
operationId: createInstanceDatatableRole
|
||||
tags:
|
||||
- setting
|
||||
@@ -1655,6 +1722,8 @@ paths:
|
||||
properties:
|
||||
name:
|
||||
type: string
|
||||
cluster:
|
||||
$ref: "#/components/schemas/DatatableRoleCluster"
|
||||
responses:
|
||||
"200":
|
||||
description: the created role
|
||||
@@ -5267,7 +5336,7 @@ paths:
|
||||
type: string
|
||||
resource_type:
|
||||
type: string
|
||||
enum: [postgres, instance]
|
||||
enum: [postgres, instance, external_instance]
|
||||
resource_path:
|
||||
type: string
|
||||
governing_workspace_id:
|
||||
@@ -33696,9 +33765,17 @@ components:
|
||||
last_setup:
|
||||
$ref: "#/components/schemas/ExternalInstancePgSetupReport"
|
||||
|
||||
DatatableRoleCluster:
|
||||
type: string
|
||||
description: >-
|
||||
The Windmill-managed Postgres cluster a data table role is a login on: Windmill's own
|
||||
(behind `instance` data tables) or the external instance cluster (behind
|
||||
`external_instance` ones). Defaults to `instance`.
|
||||
enum: [instance, external_instance]
|
||||
|
||||
InstanceDatatableRole:
|
||||
type: object
|
||||
required: [id, name, enabled]
|
||||
required: [id, name, enabled, cluster]
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
@@ -33706,6 +33783,8 @@ components:
|
||||
type: string
|
||||
enabled:
|
||||
type: boolean
|
||||
cluster:
|
||||
$ref: "#/components/schemas/DatatableRoleCluster"
|
||||
|
||||
DatatableRoleTenants:
|
||||
type: object
|
||||
@@ -33727,8 +33806,10 @@ components:
|
||||
supported:
|
||||
type: boolean
|
||||
description: >-
|
||||
Whether this data table can be put under roles at all. Only one backed by the
|
||||
instance database can: a role is a login on that cluster.
|
||||
Whether this data table can be put under roles at all. Only one on a database Windmill
|
||||
manages can: a role is a login on that database's cluster.
|
||||
cluster:
|
||||
$ref: "#/components/schemas/DatatableRoleCluster"
|
||||
permissioned:
|
||||
type: boolean
|
||||
default_role:
|
||||
@@ -36002,6 +36083,7 @@ components:
|
||||
- postgresql
|
||||
- mysql
|
||||
- instance
|
||||
- external_instance
|
||||
resource_path:
|
||||
type: string
|
||||
required:
|
||||
@@ -36065,6 +36147,7 @@ components:
|
||||
enum:
|
||||
- postgresql
|
||||
- instance
|
||||
- external_instance
|
||||
resource_path:
|
||||
type: string
|
||||
required:
|
||||
|
||||
@@ -6,22 +6,67 @@
|
||||
* LICENSE-AGPL for a copy of the license.
|
||||
*/
|
||||
|
||||
//! The instance's data table role catalog.
|
||||
//! The instance's data table role catalogs.
|
||||
//!
|
||||
//! A data table role is a real Postgres login role on the Windmill cluster, named exactly as the
|
||||
//! user named it, shared by every instance database. Windmill decides who may ask for a role (the
|
||||
//! per-data-table tenant lists in [`crate::workspaces`]); Postgres decides what the role may then
|
||||
//! touch. The catalog here is only the first half's vocabulary plus the cluster provisioning.
|
||||
//! A data table role is a real Postgres login role on one cluster — Windmill's own, or the external
|
||||
//! instance cluster — named exactly as the user named it, shared by every database Windmill manages
|
||||
//! on that cluster. Each cluster has its own catalog: a role exists where it was created and nowhere
|
||||
//! else. Windmill decides who may ask for a role (the per-data-table tenant lists in
|
||||
//! [`crate::workspaces`]); Postgres decides what the role may then touch. The catalog here is only
|
||||
//! the first half's vocabulary plus the cluster provisioning.
|
||||
//!
|
||||
//! Entries are keyed by a generated id so a rename moves nothing else: tenants name the id.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::{
|
||||
error::{Error, Result},
|
||||
workspaces::DataTableCatalogResourceType,
|
||||
DB,
|
||||
};
|
||||
|
||||
/// The cluster a role catalog belongs to.
|
||||
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum DatatableRoleCluster {
|
||||
/// Windmill's own Postgres, behind `instance` data tables.
|
||||
#[default]
|
||||
Instance,
|
||||
/// The external instance cluster, behind `external_instance` data tables.
|
||||
ExternalInstance,
|
||||
}
|
||||
|
||||
impl DatatableRoleCluster {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Instance => "instance",
|
||||
Self::ExternalInstance => "external_instance",
|
||||
}
|
||||
}
|
||||
|
||||
pub fn parse(value: &str) -> Result<Self> {
|
||||
match value {
|
||||
"instance" => Ok(Self::Instance),
|
||||
"external_instance" => Ok(Self::ExternalInstance),
|
||||
other => Err(Error::BadRequest(format!(
|
||||
"Unknown data table role cluster '{other}': expected instance or external_instance"
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
/// The cluster whose roles a data table on `kind` can use. `None` for a resource-backed one,
|
||||
/// which is never under roles.
|
||||
pub fn of(kind: DataTableCatalogResourceType) -> Option<Self> {
|
||||
match kind {
|
||||
DataTableCatalogResourceType::Instance => Some(Self::Instance),
|
||||
DataTableCatalogResourceType::ExternalInstance => Some(Self::ExternalInstance),
|
||||
DataTableCatalogResourceType::Postgresql => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The connection every data table resolved to before roles existed (`custom_instance_user`). It
|
||||
/// owns every pre-existing object, so it is a reserved name rather than a catalog entry: never
|
||||
/// created, renamed or dropped.
|
||||
@@ -164,28 +209,42 @@ pub async fn lock_instance_databases_governance<'a>(
|
||||
/// need the names — but callers MUST NOT let `pwd` reach a response, a log line, an audit record
|
||||
/// or an export. Nothing about who may call it: the credential is the whole risk, and `Debug` is
|
||||
/// hand-written to redact it for the same reason.
|
||||
pub async fn read_role_catalog(db: &DB) -> Result<DatatableRoleCatalog> {
|
||||
crate::datatable_roles_oss::read_role_catalog(db).await
|
||||
pub async fn read_role_catalog(
|
||||
db: &DB,
|
||||
cluster: DatatableRoleCluster,
|
||||
) -> Result<DatatableRoleCatalog> {
|
||||
crate::datatable_roles_oss::read_role_catalog(db, cluster).await
|
||||
}
|
||||
|
||||
/// As [`read_role_catalog`], reading inside the caller's transaction so the value is the one
|
||||
/// [`lock_role_catalog`] is protecting. Same disclosure contract.
|
||||
pub async fn read_role_catalog_tx(
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
cluster: DatatableRoleCluster,
|
||||
) -> Result<DatatableRoleCatalog> {
|
||||
crate::datatable_roles_oss::read_role_catalog_tx(tx).await
|
||||
crate::datatable_roles_oss::read_role_catalog_tx(tx, cluster).await
|
||||
}
|
||||
|
||||
/// Record a role, in the caller's transaction so it commits with the `CREATE ROLE` it describes.
|
||||
/// The cluster a role belongs to, or `None` if no role has this id.
|
||||
pub async fn role_cluster(
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
id: &str,
|
||||
) -> Result<Option<DatatableRoleCluster>> {
|
||||
crate::datatable_roles_oss::role_cluster(tx, id).await
|
||||
}
|
||||
|
||||
/// Record a role, in the caller's transaction. On Windmill's own cluster that commits it with the
|
||||
/// `CREATE ROLE` it describes; on the external cluster the role already exists by then.
|
||||
///
|
||||
/// Authorization: writes a generated Postgres credential. Callers MUST restrict this to superadmin
|
||||
/// paths and MUST hold [`lock_role_catalog`] on `tx`.
|
||||
pub async fn insert_role_catalog_entry(
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
id: &str,
|
||||
cluster: DatatableRoleCluster,
|
||||
role: &InstanceDatatableRole,
|
||||
) -> Result<()> {
|
||||
crate::datatable_roles_oss::insert_role_catalog_entry(tx, id, role).await
|
||||
crate::datatable_roles_oss::insert_role_catalog_entry(tx, id, cluster, role).await
|
||||
}
|
||||
|
||||
/// Update a role's recorded name, login flag and password. Same contract as
|
||||
@@ -215,7 +274,7 @@ pub fn role_id_by_name<'a>(catalog: &'a DatatableRoleCatalog, name: &str) -> Res
|
||||
.find(|(_, role)| role.name == name)
|
||||
.ok_or_else(|| {
|
||||
Error::NotFound(format!(
|
||||
"'{name}' is not a data table role of this instance. Defined roles: {}.",
|
||||
"'{name}' is not a data table role of this database's cluster. Defined roles: {}.",
|
||||
catalog
|
||||
.values()
|
||||
.map(|r| r.name.as_str())
|
||||
@@ -231,67 +290,87 @@ pub fn role_id_by_name<'a>(catalog: &'a DatatableRoleCatalog, name: &str) -> Res
|
||||
Ok(entry.0.as_str())
|
||||
}
|
||||
|
||||
/// 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.
|
||||
pub async fn registered_instance_databases(db: &DB) -> Result<Vec<String>> {
|
||||
crate::datatable_roles_oss::registered_instance_databases(db).await
|
||||
/// Every database Windmill manages on `cluster`. Role provisioning has to reach all of them: a role
|
||||
/// that cannot `CONNECT` to a database is refused by Postgres before any grant matters.
|
||||
pub async fn registered_instance_databases(
|
||||
db: &DB,
|
||||
cluster: DatatableRoleCluster,
|
||||
) -> Result<Vec<String>> {
|
||||
crate::datatable_roles_oss::registered_instance_databases(db, cluster).await
|
||||
}
|
||||
|
||||
/// `CONNECT` on `dbname` for every enabled role, and none for `PUBLIC`. Run at role creation, at
|
||||
/// database creation, and lazily whenever an instance data table is administered, so a database
|
||||
/// provisioned before a role existed is repaired rather than left silently unreachable.
|
||||
/// `CONNECT` on `dbname` for every enabled role of `cluster`, and none for `PUBLIC`. Run at role
|
||||
/// creation, at database creation, and lazily whenever a managed data table is administered, so a
|
||||
/// database provisioned before a role existed is repaired rather than left silently unreachable.
|
||||
///
|
||||
/// Authorization: rewrites a database's ACL with the server's own credentials and checks nothing.
|
||||
/// Callers MUST have authorized administration of `dbname` — superadmin, or an admin of the
|
||||
/// workspace governing a data table on it.
|
||||
pub async fn converge_connect_grants(db: &DB, dbname: &str) -> Result<()> {
|
||||
crate::datatable_roles_oss::converge_connect_grants(db, dbname).await
|
||||
pub async fn converge_connect_grants(
|
||||
db: &DB,
|
||||
cluster: DatatableRoleCluster,
|
||||
dbname: &str,
|
||||
) -> Result<()> {
|
||||
crate::datatable_roles_oss::converge_connect_grants(db, cluster, dbname).await
|
||||
}
|
||||
|
||||
/// As [`converge_connect_grants`], with a catalog the caller already read. Same contract.
|
||||
/// As [`converge_connect_grants`], with the catalog of `cluster` the caller already read. Same
|
||||
/// contract.
|
||||
pub async fn converge_connect_grants_with(
|
||||
db: &DB,
|
||||
cluster: DatatableRoleCluster,
|
||||
dbname: &str,
|
||||
catalog: &DatatableRoleCatalog,
|
||||
) -> Result<()> {
|
||||
crate::datatable_roles_oss::converge_connect_grants_with(db, dbname, catalog).await
|
||||
crate::datatable_roles_oss::converge_connect_grants_with(db, cluster, dbname, catalog).await
|
||||
}
|
||||
|
||||
/// `CREATE ROLE <name> LOGIN PASSWORD ...; GRANT <name> TO custom_instance_user`, and `CONNECT` on
|
||||
/// every registered database. No privileges beyond that — an admin grants them through SQL or the
|
||||
/// ACL editor.
|
||||
/// `CREATE ROLE <name> LOGIN PASSWORD ...; GRANT <name> TO custom_instance_user` on `cluster`. No
|
||||
/// privileges beyond that — an admin grants them through SQL or the ACL editor.
|
||||
///
|
||||
/// On Windmill's own cluster the DDL runs on `tx`, so it commits with the catalog row. The external
|
||||
/// cluster is another server: the role is created there before `tx` commits, and callers MUST drop
|
||||
/// it again ([`drop_datatable_role`]) if `tx` then fails to commit.
|
||||
///
|
||||
/// Authorization: creates a cluster-wide Postgres login. Callers MUST restrict this to superadmin
|
||||
/// paths, and MUST hold [`lock_role_catalog`] on the same transaction.
|
||||
pub async fn create_instance_role(
|
||||
/// paths, and MUST hold [`lock_role_catalog`] on `tx`.
|
||||
pub async fn create_datatable_role(
|
||||
db: &DB,
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
cluster: DatatableRoleCluster,
|
||||
name: &str,
|
||||
password: &str,
|
||||
) -> Result<()> {
|
||||
crate::datatable_roles_oss::create_instance_role(tx, name, password).await
|
||||
crate::datatable_roles_oss::create_datatable_role(db, tx, cluster, name, password).await
|
||||
}
|
||||
|
||||
/// Authorization: alters a cluster-wide Postgres login. Callers MUST restrict this to superadmin
|
||||
/// paths, and MUST hold [`lock_role_catalog`] on the same transaction.
|
||||
pub async fn set_instance_role_login(
|
||||
/// paths, and MUST hold [`lock_role_catalog`] on `tx`.
|
||||
pub async fn set_datatable_role_login(
|
||||
db: &DB,
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
cluster: DatatableRoleCluster,
|
||||
name: &str,
|
||||
enabled: bool,
|
||||
) -> Result<()> {
|
||||
crate::datatable_roles_oss::set_instance_role_login(tx, name, enabled).await
|
||||
crate::datatable_roles_oss::set_datatable_role_login(db, tx, cluster, name, enabled).await
|
||||
}
|
||||
|
||||
/// A rename discards an md5-hashed password, so the caller has to hand over a fresh one.
|
||||
/// A rename discards an md5-hashed password, so the caller has to hand over a fresh one. On the
|
||||
/// external cluster the rename lands before `tx` commits, and callers MUST rename it back if `tx`
|
||||
/// then fails to commit.
|
||||
///
|
||||
/// Authorization: renames a cluster-wide Postgres login. Callers MUST restrict this to superadmin
|
||||
/// paths, and MUST hold [`lock_role_catalog`] on the same transaction.
|
||||
pub async fn rename_instance_role(
|
||||
/// paths, and MUST hold [`lock_role_catalog`] on `tx`.
|
||||
pub async fn rename_datatable_role(
|
||||
db: &DB,
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
cluster: DatatableRoleCluster,
|
||||
from: &str,
|
||||
to: &str,
|
||||
password: &str,
|
||||
) -> Result<()> {
|
||||
crate::datatable_roles_oss::rename_instance_role(tx, from, to, password).await
|
||||
crate::datatable_roles_oss::rename_datatable_role(db, tx, cluster, from, to, password).await
|
||||
}
|
||||
|
||||
/// A role owning anything in any database blocks its own `DROP ROLE`, and both its objects and the
|
||||
@@ -299,8 +378,9 @@ pub async fn rename_instance_role(
|
||||
/// registry. An unreachable database aborts the whole delete: dropping the role while one database
|
||||
/// still holds objects owned by it leaves those objects owned by a numeric OID nobody can name.
|
||||
///
|
||||
/// Each pass runs as the instance's own Postgres user rather than `custom_instance_user`, which
|
||||
/// owns the databases and can therefore revoke a grant whoever made it. `custom_instance_user`
|
||||
/// Each pass runs as the cluster's administrator rather than `custom_instance_user`: on Windmill's
|
||||
/// own cluster the instance's Postgres user, on the external one its configured admin login. Both
|
||||
/// own the databases and can therefore revoke a grant whoever made it. `custom_instance_user`
|
||||
/// could only undo what it granted itself, so a privilege planted by an operator in psql — the
|
||||
/// ordinary way privileges reach a role — would survive and block the drop.
|
||||
///
|
||||
@@ -308,16 +388,19 @@ pub async fn rename_instance_role(
|
||||
/// MUST restrict this to superadmin paths, and MUST hold [`lock_role_catalog`] on `tx`.
|
||||
///
|
||||
/// The per-database passes open their own connections and cannot join `tx`; the lock is what keeps
|
||||
/// a concurrent mutation out while they run. Only the final `DROP ROLE` is on `tx`, so it commits
|
||||
/// or rolls back with the catalog write that forgets the role. Those passes commit as they go, so
|
||||
/// callers MUST have disabled the role in an earlier committed transaction: a failure part-way
|
||||
/// then leaves a disabled role to retry, not an enabled one already stripped in some databases.
|
||||
pub async fn drop_instance_role(
|
||||
/// a concurrent mutation out while they run. On Windmill's own cluster only the final `DROP ROLE`
|
||||
/// is on `tx`, so it commits or rolls back with the catalog write that forgets the role; on the
|
||||
/// external cluster it runs there, and tolerates a role already gone so a retry after a failed
|
||||
/// commit can finish. The passes commit as they go, so callers MUST have disabled the role in an
|
||||
/// earlier committed transaction: a failure part-way then leaves a disabled role to retry, not an
|
||||
/// enabled one already stripped in some databases.
|
||||
pub async fn drop_datatable_role(
|
||||
db: &DB,
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
cluster: DatatableRoleCluster,
|
||||
name: &str,
|
||||
) -> Result<()> {
|
||||
crate::datatable_roles_oss::drop_instance_role(db, tx, name).await
|
||||
crate::datatable_roles_oss::drop_datatable_role(db, tx, cluster, name).await
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -24,12 +24,12 @@ pub fn datatable_roles_unavailable() -> Error {
|
||||
#[cfg(all(feature = "private", feature = "enterprise"))]
|
||||
pub(crate) use crate::datatable_roles_ee::{
|
||||
can_use_datatable_role, can_use_datatable_role_in_governing_workspace, converge_connect_grants,
|
||||
converge_connect_grants_with, create_instance_role, delete_role_catalog_entry,
|
||||
drop_instance_role, ensure_can_use_datatable_role, ensure_datatable_admin_access,
|
||||
converge_connect_grants_with, create_datatable_role, delete_role_catalog_entry,
|
||||
drop_datatable_role, ensure_can_use_datatable_role, ensure_datatable_admin_access,
|
||||
ensure_instance_db_grant_options_unchecked, forget_datatable_role_everywhere,
|
||||
insert_role_catalog_entry, read_role_catalog, read_role_catalog_tx,
|
||||
registered_instance_databases, rename_instance_role, resolve_datatable_role_connection,
|
||||
set_instance_role_login, update_role_catalog_entry,
|
||||
registered_instance_databases, rename_datatable_role, resolve_datatable_role_connection,
|
||||
role_cluster, set_datatable_role_login, update_role_catalog_entry,
|
||||
};
|
||||
|
||||
#[cfg(not(all(feature = "private", feature = "enterprise")))]
|
||||
@@ -39,7 +39,7 @@ pub(crate) use ce::*;
|
||||
mod ce {
|
||||
use super::datatable_roles_unavailable as unavailable;
|
||||
use crate::{
|
||||
datatable_roles::{DatatableRoleCatalog, InstanceDatatableRole},
|
||||
datatable_roles::{DatatableRoleCatalog, DatatableRoleCluster, InstanceDatatableRole},
|
||||
db::AuthedRef,
|
||||
error::Result,
|
||||
workspaces::{
|
||||
@@ -50,17 +50,31 @@ mod ce {
|
||||
|
||||
type Tx<'a> = sqlx::Transaction<'a, sqlx::Postgres>;
|
||||
|
||||
pub(crate) async fn read_role_catalog(_db: &DB) -> Result<DatatableRoleCatalog> {
|
||||
pub(crate) async fn read_role_catalog(
|
||||
_db: &DB,
|
||||
_cluster: DatatableRoleCluster,
|
||||
) -> Result<DatatableRoleCatalog> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn read_role_catalog_tx(_tx: &mut Tx<'_>) -> Result<DatatableRoleCatalog> {
|
||||
pub(crate) async fn read_role_catalog_tx(
|
||||
_tx: &mut Tx<'_>,
|
||||
_cluster: DatatableRoleCluster,
|
||||
) -> Result<DatatableRoleCatalog> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn role_cluster(
|
||||
_tx: &mut Tx<'_>,
|
||||
_id: &str,
|
||||
) -> Result<Option<DatatableRoleCluster>> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn insert_role_catalog_entry(
|
||||
_tx: &mut Tx<'_>,
|
||||
_id: &str,
|
||||
_cluster: DatatableRoleCluster,
|
||||
_role: &InstanceDatatableRole,
|
||||
) -> Result<()> {
|
||||
Err(unavailable())
|
||||
@@ -78,43 +92,57 @@ mod ce {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn registered_instance_databases(_db: &DB) -> Result<Vec<String>> {
|
||||
pub(crate) async fn registered_instance_databases(
|
||||
_db: &DB,
|
||||
_cluster: DatatableRoleCluster,
|
||||
) -> Result<Vec<String>> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
/// Nothing to converge: with no roles to admit, an instance database keeps the `CONNECT`
|
||||
/// grants it was created with, `PUBLIC`'s included, as it did before roles existed.
|
||||
pub(crate) async fn converge_connect_grants(_db: &DB, _dbname: &str) -> Result<()> {
|
||||
/// Nothing to converge: with no roles to admit, a managed database keeps the `CONNECT` grants
|
||||
/// it was created with, as it did before roles existed.
|
||||
pub(crate) async fn converge_connect_grants(
|
||||
_db: &DB,
|
||||
_cluster: DatatableRoleCluster,
|
||||
_dbname: &str,
|
||||
) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// As [`converge_connect_grants`].
|
||||
pub(crate) async fn converge_connect_grants_with(
|
||||
_db: &DB,
|
||||
_cluster: DatatableRoleCluster,
|
||||
_dbname: &str,
|
||||
_catalog: &DatatableRoleCatalog,
|
||||
) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn create_instance_role(
|
||||
pub(crate) async fn create_datatable_role(
|
||||
_db: &DB,
|
||||
_tx: &mut Tx<'_>,
|
||||
_cluster: DatatableRoleCluster,
|
||||
_name: &str,
|
||||
_password: &str,
|
||||
) -> Result<()> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn set_instance_role_login(
|
||||
pub(crate) async fn set_datatable_role_login(
|
||||
_db: &DB,
|
||||
_tx: &mut Tx<'_>,
|
||||
_cluster: DatatableRoleCluster,
|
||||
_name: &str,
|
||||
_enabled: bool,
|
||||
) -> Result<()> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn rename_instance_role(
|
||||
pub(crate) async fn rename_datatable_role(
|
||||
_db: &DB,
|
||||
_tx: &mut Tx<'_>,
|
||||
_cluster: DatatableRoleCluster,
|
||||
_from: &str,
|
||||
_to: &str,
|
||||
_password: &str,
|
||||
@@ -122,12 +150,18 @@ mod ce {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn drop_instance_role(_db: &DB, _tx: &mut Tx<'_>, _name: &str) -> Result<()> {
|
||||
pub(crate) async fn drop_datatable_role(
|
||||
_db: &DB,
|
||||
_tx: &mut Tx<'_>,
|
||||
_cluster: DatatableRoleCluster,
|
||||
_name: &str,
|
||||
) -> Result<()> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn ensure_instance_db_grant_options_unchecked(
|
||||
_db: &DB,
|
||||
_cluster: DatatableRoleCluster,
|
||||
_dbname: &str,
|
||||
) -> Result<()> {
|
||||
Err(unavailable())
|
||||
|
||||
@@ -17,7 +17,7 @@
|
||||
//! The cluster may hold data Windmill did not create. Two Windmill instances sharing one is not
|
||||
//! supported: each would keep resetting the passwords the other depends on.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
@@ -124,42 +124,281 @@ pub async fn external_instance_pg_status(db: &DB) -> Result<ExternalInstancePgSt
|
||||
})
|
||||
}
|
||||
|
||||
/// Refuse to unset the cluster while Windmill still has databases on it: every data table and
|
||||
/// Ducklake catalog there would stop resolving. Allowed on every edition, so a downgraded
|
||||
/// instance can still clear a setting it no longer uses.
|
||||
/// The databases Windmill created on the external cluster, without the passwords kept beside them.
|
||||
pub async fn external_instance_databases(db: &DB) -> Result<BTreeMap<String, CustomInstanceDb>> {
|
||||
Ok(read_external_instance_pg_state(db).await?.databases)
|
||||
}
|
||||
|
||||
/// The workspaces whose data tables or Ducklake catalogs name each database on the external cluster,
|
||||
/// and the forks whose Ducklake namespaces there are still waiting to be cleaned up: those rows
|
||||
/// outlive a settings change, and cleanup cannot drop a namespace in a database that is gone.
|
||||
///
|
||||
/// Authorization: reads every workspace's settings and checks nothing. Callers MUST be superadmin
|
||||
/// or an internal lifecycle path.
|
||||
pub async fn external_instance_database_usages<'c>(
|
||||
db: impl sqlx::PgExecutor<'c>,
|
||||
) -> Result<BTreeMap<String, BTreeSet<String>>> {
|
||||
let rows = sqlx::query_as::<_, (String, String)>(
|
||||
"SELECT ws.workspace_id, entry->'database'->>'resource_path'
|
||||
FROM workspace_settings ws
|
||||
CROSS JOIN LATERAL jsonb_each(
|
||||
CASE WHEN jsonb_typeof(ws.datatable->'datatables') = 'object'
|
||||
THEN ws.datatable->'datatables'
|
||||
ELSE '{}'::jsonb END
|
||||
) AS dt(k, entry)
|
||||
WHERE entry->'database'->>'resource_type' = 'external_instance'
|
||||
AND entry->'database'->>'resource_path' IS NOT NULL
|
||||
UNION ALL
|
||||
SELECT ws.workspace_id, entry->'catalog'->>'resource_path'
|
||||
FROM workspace_settings ws
|
||||
CROSS JOIN LATERAL jsonb_each(
|
||||
CASE WHEN jsonb_typeof(ws.ducklake->'ducklakes') = 'object'
|
||||
THEN ws.ducklake->'ducklakes'
|
||||
ELSE '{}'::jsonb END
|
||||
) AS dl(k, entry)
|
||||
WHERE entry->'catalog'->>'resource_type' = 'external_instance'
|
||||
AND entry->'catalog'->>'resource_path' IS NOT NULL
|
||||
UNION ALL
|
||||
SELECT workspace_id, substring(catalog FROM length('external_instance:') + 1)
|
||||
FROM fork_ducklake_namespace
|
||||
WHERE catalog LIKE 'external\\_instance:%'",
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
let mut usages: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
|
||||
for (workspace_id, dbname) in rows {
|
||||
usages.entry(dbname).or_default().insert(workspace_id);
|
||||
}
|
||||
Ok(usages)
|
||||
}
|
||||
|
||||
/// Refuse to unset the cluster while Windmill still has databases or data table roles on it, or a
|
||||
/// workspace still points at one: every data table there would stop resolving, and every role
|
||||
/// would be a login nothing can drop any more. Allowed on every edition, so a
|
||||
/// downgraded instance can still clear a setting it no longer uses.
|
||||
pub async fn ensure_external_instance_pg_removable(db: &DB) -> Result<()> {
|
||||
ensure_external_instance_pg_unused(db, &format!("removing {EXTERNAL_INSTANCE_PG_SETTING}"))
|
||||
.await
|
||||
}
|
||||
|
||||
/// Refuse while Windmill has databases or data table roles on the cluster, or a workspace points
|
||||
/// at one of its databases. `before` finishes the sentence saying what to do first.
|
||||
async fn ensure_external_instance_pg_unused(db: &DB, before: &str) -> Result<()> {
|
||||
let state = read_external_instance_pg_state(db).await?;
|
||||
if state.databases.is_empty() {
|
||||
let usages = external_instance_database_usages(db).await?;
|
||||
let roles = sqlx::query_scalar::<_, String>(
|
||||
"SELECT name FROM datatable_role WHERE cluster = 'external_instance' ORDER BY name",
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
if state.databases.is_empty() && usages.is_empty() && roles.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let names = state
|
||||
.databases
|
||||
.keys()
|
||||
.cloned()
|
||||
.collect::<Vec<_>>()
|
||||
.join(", ");
|
||||
let mut held = vec![];
|
||||
if !(state.databases.is_empty() && usages.is_empty()) {
|
||||
let names = state
|
||||
.databases
|
||||
.keys()
|
||||
.chain(usages.keys())
|
||||
.collect::<BTreeSet<_>>()
|
||||
.into_iter()
|
||||
.cloned()
|
||||
.collect::<Vec<_>>()
|
||||
.join(", ");
|
||||
held.push(format!("databases in use ({names})"));
|
||||
}
|
||||
if !roles.is_empty() {
|
||||
held.push(format!("data table roles ({})", roles.join(", ")));
|
||||
}
|
||||
Err(Error::BadRequest(format!(
|
||||
"The external instance cluster still holds databases Windmill created ({names}). Drop \
|
||||
them before removing {EXTERNAL_INSTANCE_PG_SETTING}."
|
||||
"The external instance cluster still holds {}. Drop them and repoint the data tables and \
|
||||
Ducklake catalogs using them before {before}.",
|
||||
held.join(" and ")
|
||||
)))
|
||||
}
|
||||
|
||||
/// Check a write to [`EXTERNAL_INSTANCE_PG_SETTING`] before it happens: `None`, null or an empty
|
||||
/// string unsets it. Every writer of global settings calls this, the per-key and bulk endpoints
|
||||
/// as well as the declarative sync.
|
||||
pub async fn check_external_instance_pg_write(
|
||||
/// Refuse a workspace setting that newly names an `external_instance` database on an edition
|
||||
/// without them.
|
||||
pub fn ensure_external_instance_available() -> Result<()> {
|
||||
crate::external_instance_pg_oss::ensure_external_instance_available()
|
||||
}
|
||||
|
||||
/// The connection an `external_instance` database resolves to: `custom_instance_user`, or the
|
||||
/// replication user, on the external cluster.
|
||||
///
|
||||
/// Authorization: returns live credentials and checks nothing. Callers MUST have authorized access
|
||||
/// to the data table that names `dbname`.
|
||||
pub async fn external_instance_connection_unchecked(
|
||||
db: &DB,
|
||||
dbname: &str,
|
||||
replication: bool,
|
||||
) -> Result<crate::PgDatabase> {
|
||||
crate::external_instance_pg_oss::external_instance_connection_unchecked(db, dbname, replication)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Create `dbname` on the external cluster and register it. Refuses a name already taken there,
|
||||
/// whoever took it.
|
||||
///
|
||||
/// Authorization: checks nothing. Callers MUST be superadmin, or be cloning a data table they may
|
||||
/// fork into a `wm_fork_` database.
|
||||
pub async fn create_external_instance_database_unchecked(
|
||||
db: &DB,
|
||||
dbname: &str,
|
||||
tag: &str,
|
||||
) -> Result<()> {
|
||||
crate::external_instance_pg_oss::create_external_instance_database_unchecked(db, dbname, tag)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Drop `dbname` from the external cluster: only a database Windmill registered creating, and still
|
||||
/// carries the mark it set there. Refused while a data table names it, except one in
|
||||
/// `usage_allowed_in`: the fork whose own copy is being cleaned up.
|
||||
///
|
||||
/// Authorization: checks nothing. Callers MUST be superadmin, or be deleting the fork that owns
|
||||
/// this `wm_fork_` database.
|
||||
pub async fn drop_external_instance_database_unchecked(
|
||||
db: &DB,
|
||||
dbname: &str,
|
||||
usage_allowed_in: Option<&str>,
|
||||
) -> Result<()> {
|
||||
crate::external_instance_pg_oss::drop_external_instance_database_unchecked(
|
||||
db,
|
||||
dbname,
|
||||
usage_allowed_in,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Serializes everything that changes which databases exist on the external cluster, or which data
|
||||
/// tables name them: setup, creates, drops, and data table saves. Held until `tx` ends.
|
||||
pub async fn lock_external_instance_pg_state(
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
) -> Result<()> {
|
||||
sqlx::query("SELECT pg_advisory_xact_lock(hashtext($1))")
|
||||
.bind(EXTERNAL_INSTANCE_PG_STATE_SETTING)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Refuse a data table naming `dbname` unless Windmill created it on the external cluster. Takes
|
||||
/// the lock drops take, so none can remove the database before `tx`, which saves the data table,
|
||||
/// commits.
|
||||
pub async fn ensure_external_instance_database_registered(
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
dbname: &str,
|
||||
) -> Result<()> {
|
||||
lock_external_instance_pg_state(tx).await?;
|
||||
if read_external_instance_pg_state(&mut **tx)
|
||||
.await?
|
||||
.databases
|
||||
.contains_key(dbname)
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
Err(Error::BadRequest(format!(
|
||||
"Windmill did not create a database named '{dbname}' on the external instance cluster. \
|
||||
Create it from the instance settings first."
|
||||
)))
|
||||
}
|
||||
|
||||
/// Write [`EXTERNAL_INSTANCE_PG_SETTING`]: `None`, null or an empty string unsets it. Every writer
|
||||
/// of global settings goes through this for that key — the per-key and bulk endpoints as well as
|
||||
/// the declarative sync — instead of writing the row itself.
|
||||
///
|
||||
/// The checks and the write share one transaction holding [`lock_external_instance_pg_state`]. A
|
||||
/// check taken outside it could pass while a database create still reads the old cluster, which
|
||||
/// would then register a database there after the setting names another one.
|
||||
///
|
||||
/// Authorization: checks nothing. Callers MUST be superadmin, or the declarative instance config
|
||||
/// sync, which applies what the operator deployed.
|
||||
pub async fn write_external_instance_pg_setting(
|
||||
db: &DB,
|
||||
value: Option<&serde_json::Value>,
|
||||
) -> Result<()> {
|
||||
let value = match value {
|
||||
None | Some(serde_json::Value::Null) => None,
|
||||
Some(serde_json::Value::String(s)) if s.trim().is_empty() => None,
|
||||
Some(value) => Some(value),
|
||||
};
|
||||
let mut tx = db.begin().await?;
|
||||
lock_external_instance_pg_state(&mut tx).await?;
|
||||
match value {
|
||||
None | Some(serde_json::Value::Null) => ensure_external_instance_pg_removable(db).await,
|
||||
Some(serde_json::Value::String(s)) if s.trim().is_empty() => {
|
||||
ensure_external_instance_pg_removable(db).await
|
||||
None => {
|
||||
ensure_external_instance_pg_removable(db).await?;
|
||||
sqlx::query("DELETE FROM global_settings WHERE name = $1")
|
||||
.bind(EXTERNAL_INSTANCE_PG_SETTING)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
}
|
||||
Some(value) => {
|
||||
crate::external_instance_pg_oss::validate_external_instance_pg_setting(value)
|
||||
crate::external_instance_pg_oss::validate_external_instance_pg_setting(value)?;
|
||||
ensure_external_instance_pg_not_repointed(db, value).await?;
|
||||
sqlx::query(
|
||||
"INSERT INTO global_settings (name, value) VALUES ($1, $2)
|
||||
ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value, updated_at = now()",
|
||||
)
|
||||
.bind(EXTERNAL_INSTANCE_PG_SETTING)
|
||||
.bind(value)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
tx.commit().await?;
|
||||
tracing::info!(
|
||||
"{} global setting {EXTERNAL_INSTANCE_PG_SETTING}",
|
||||
if value.is_some() { "Set" } else { "Unset" }
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// [`write_external_instance_pg_setting`] for a settings diff: writes the key if the diff touches
|
||||
/// it, and takes it out of the diff so the generic apply does not write it again.
|
||||
///
|
||||
/// Authorization: checks nothing. Callers MUST be superadmin, or the declarative instance config
|
||||
/// sync, which applies what the operator deployed.
|
||||
pub async fn write_external_instance_pg_from_diff(
|
||||
db: &DB,
|
||||
diff: &mut crate::instance_config::SettingsDiff,
|
||||
) -> Result<()> {
|
||||
if let Some(value) = diff.upserts.remove(EXTERNAL_INSTANCE_PG_SETTING) {
|
||||
write_external_instance_pg_setting(db, Some(&value)).await?;
|
||||
}
|
||||
if let Some(i) = diff
|
||||
.deletes
|
||||
.iter()
|
||||
.position(|k| k == EXTERNAL_INSTANCE_PG_SETTING)
|
||||
{
|
||||
diff.deletes.remove(i);
|
||||
write_external_instance_pg_setting(db, None).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Refuse pointing the setting at another host or port while databases or data table roles live on
|
||||
/// the current one. Data tables name databases, and the role catalog names logins, not clusters, so
|
||||
/// both would silently resolve to whatever the new cluster holds under the same names. Other fields
|
||||
/// (admin login, sslmode) may change freely.
|
||||
async fn ensure_external_instance_pg_not_repointed(
|
||||
db: &DB,
|
||||
value: &serde_json::Value,
|
||||
) -> Result<()> {
|
||||
let Some(current) = read_external_instance_pg_config(db).await? else {
|
||||
return Ok(());
|
||||
};
|
||||
let Ok(desired) = serde_json::from_value::<ExternalInstancePg>(value.clone()) else {
|
||||
return Ok(());
|
||||
};
|
||||
let address = |c: &ExternalInstancePg| (c.host.trim().to_lowercase(), c.port.unwrap_or(5432));
|
||||
if address(¤t) == address(&desired) {
|
||||
return Ok(());
|
||||
}
|
||||
ensure_external_instance_pg_unused(
|
||||
db,
|
||||
&format!("pointing {EXTERNAL_INSTANCE_PG_SETTING} at another cluster"),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Converge the external cluster on the configured login: check what it can do, create or update
|
||||
|
||||
@@ -19,25 +19,62 @@ pub fn external_instance_pg_unavailable() -> Error {
|
||||
|
||||
#[cfg(all(feature = "private", feature = "enterprise"))]
|
||||
pub(crate) use crate::external_instance_pg_ee::{
|
||||
setup_external_instance_pg_unchecked, validate_external_instance_pg_setting,
|
||||
create_external_instance_database_unchecked, drop_external_instance_database_unchecked,
|
||||
external_instance_connection_unchecked, setup_external_instance_pg_unchecked,
|
||||
validate_external_instance_pg_setting,
|
||||
};
|
||||
|
||||
#[cfg(all(feature = "private", feature = "enterprise"))]
|
||||
pub(crate) fn ensure_external_instance_available() -> crate::error::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(not(all(feature = "private", feature = "enterprise")))]
|
||||
pub(crate) use ce::*;
|
||||
|
||||
#[cfg(not(all(feature = "private", feature = "enterprise")))]
|
||||
mod ce {
|
||||
use super::external_instance_pg_unavailable as unavailable;
|
||||
use crate::{error::Result, external_instance_pg::ExternalInstancePgSetupReport, DB};
|
||||
use crate::{
|
||||
error::Result, external_instance_pg::ExternalInstancePgSetupReport, PgDatabase, DB,
|
||||
};
|
||||
|
||||
pub(crate) fn validate_external_instance_pg_setting(_value: &serde_json::Value) -> Result<()> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) fn ensure_external_instance_available() -> Result<()> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn setup_external_instance_pg_unchecked(
|
||||
_db: &DB,
|
||||
_rotate_passwords: bool,
|
||||
) -> Result<ExternalInstancePgSetupReport> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn external_instance_connection_unchecked(
|
||||
_db: &DB,
|
||||
_dbname: &str,
|
||||
_replication: bool,
|
||||
) -> Result<PgDatabase> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn create_external_instance_database_unchecked(
|
||||
_db: &DB,
|
||||
_dbname: &str,
|
||||
_tag: &str,
|
||||
) -> Result<()> {
|
||||
Err(unavailable())
|
||||
}
|
||||
|
||||
pub(crate) async fn drop_external_instance_database_unchecked(
|
||||
_db: &DB,
|
||||
_dbname: &str,
|
||||
_usage_allowed_in: Option<&str>,
|
||||
) -> Result<()> {
|
||||
Err(unavailable())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1395,14 +1395,8 @@ pub async fn sync_global_settings_declarative(
|
||||
crate::global_settings::parse_allowed_origins_setting(desired.get(origins_key))
|
||||
.map_err(|e| anyhow::anyhow!("{origins_key}: {e}"))?;
|
||||
|
||||
let diff = diff_global_settings(current, desired, ApplyMode::Replace);
|
||||
let external_pg_key = crate::global_settings::EXTERNAL_INSTANCE_PG_SETTING;
|
||||
if diff.deletes.iter().any(|k| k == external_pg_key) {
|
||||
crate::external_instance_pg::check_external_instance_pg_write(db, None).await?;
|
||||
}
|
||||
if let Some(value) = diff.upserts.get(external_pg_key) {
|
||||
crate::external_instance_pg::check_external_instance_pg_write(db, Some(value)).await?;
|
||||
}
|
||||
let mut diff = diff_global_settings(current, desired, ApplyMode::Replace);
|
||||
crate::external_instance_pg::write_external_instance_pg_from_diff(db, &mut diff).await?;
|
||||
apply_settings_diff(db, &diff).await?;
|
||||
|
||||
Ok(())
|
||||
|
||||
@@ -1086,7 +1086,13 @@ impl PgDatabase {
|
||||
if err_str.contains("password authentication failed for user")
|
||||
&& err_str.contains("custom_instance_user")
|
||||
{
|
||||
if let Some(db) = main_db {
|
||||
// The external instance cluster has a `custom_instance_user` of its own, whose
|
||||
// password setup manages. Rotating the local one would break every instance
|
||||
// data table and fix nothing.
|
||||
let local = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
|
||||
let on_local_cluster = local.host == self.host
|
||||
&& local.port.unwrap_or(5432) == self.port.unwrap_or(5432);
|
||||
if let Some(db) = main_db.filter(|_| on_local_cluster) {
|
||||
tracing::warn!(
|
||||
"custom_instance_user password auth failed, refreshing and retrying..."
|
||||
);
|
||||
@@ -1557,18 +1563,20 @@ pub(crate) fn instance_db_grants(dbname: &str) -> String {
|
||||
)
|
||||
}
|
||||
|
||||
/// Re-apply [`instance_db_grants`] to an instance database provisioned before data table roles
|
||||
/// existed, whose grants carry no grant option. Connects as the instance's own Postgres user —
|
||||
/// the database and `public` schema owner — since only it can hand out an option it holds.
|
||||
/// Re-apply [`instance_db_grants`] to a managed database provisioned before data table roles
|
||||
/// existed, whose grants carry no grant option. Connects as the cluster's administrator — the
|
||||
/// database and `public` schema owner — since only it can hand out an option it holds.
|
||||
///
|
||||
/// Authorization: reaches an instance database with the server's own credentials and checks
|
||||
/// Authorization: reaches a managed database with the server's own credentials and checks
|
||||
/// nothing. Callers MUST have authorized administration of `dbname` — superadmin, or an admin of
|
||||
/// the workspace governing a data table on it.
|
||||
pub async fn ensure_instance_db_grant_options_unchecked(
|
||||
db: &DB,
|
||||
cluster: crate::datatable_roles::DatatableRoleCluster,
|
||||
dbname: &str,
|
||||
) -> error::Result<()> {
|
||||
crate::datatable_roles_oss::ensure_instance_db_grant_options_unchecked(db, dbname).await
|
||||
crate::datatable_roles_oss::ensure_instance_db_grant_options_unchecked(db, cluster, dbname)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Create a custom instance database: CREATE DATABASE, grant permissions, register in global_settings.
|
||||
@@ -1642,7 +1650,13 @@ pub async fn create_custom_instance_database(
|
||||
// A data table role can only reach a database it may CONNECT to, and PUBLIC's default CONNECT
|
||||
// would otherwise let every role in regardless of what this instance defines. Best-effort: a
|
||||
// failure here leaves the database usable as `admin`, and the next role change repairs it.
|
||||
if let Err(e) = crate::datatable_roles::converge_connect_grants(db, dbname).await {
|
||||
if let Err(e) = crate::datatable_roles::converge_connect_grants(
|
||||
db,
|
||||
crate::datatable_roles::DatatableRoleCluster::Instance,
|
||||
dbname,
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::warn!("Could not set CONNECT grants on instance database '{dbname}': {e}");
|
||||
}
|
||||
|
||||
@@ -1650,6 +1664,24 @@ pub async fn create_custom_instance_database(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The system's CA bundle file, for libpq clients that cannot take `sslrootcert=system`: that value
|
||||
/// needs libpq 16, and verify-full only.
|
||||
pub fn system_ca_bundle() -> Option<std::path::PathBuf> {
|
||||
std::env::var_os("SSL_CERT_FILE")
|
||||
.map(std::path::PathBuf::from)
|
||||
.into_iter()
|
||||
.chain(
|
||||
[
|
||||
"/etc/ssl/certs/ca-certificates.crt",
|
||||
"/etc/pki/tls/certs/ca-bundle.crt",
|
||||
"/etc/ssl/cert.pem",
|
||||
"/etc/ssl/ca-bundle.pem",
|
||||
]
|
||||
.map(std::path::PathBuf::from),
|
||||
)
|
||||
.find(|path| path.is_file())
|
||||
}
|
||||
|
||||
/// Connection options parsed from a database URL.
|
||||
///
|
||||
/// The only place a database URL becomes `PgConnectOptions`. Providers that mint the password
|
||||
|
||||
@@ -1397,6 +1397,18 @@ pub enum DataTableCatalogResourceType {
|
||||
#[strum(serialize = "postgres")]
|
||||
Postgresql,
|
||||
Instance,
|
||||
/// On the external instance cluster ([`crate::external_instance_pg`]). Enterprise Edition.
|
||||
#[serde(rename = "external_instance")]
|
||||
#[strum(serialize = "external_instance")]
|
||||
ExternalInstance,
|
||||
}
|
||||
|
||||
impl DataTableCatalogResourceType {
|
||||
/// A database Windmill created and administers, on its own cluster or the external one, as
|
||||
/// opposed to one a user brought as a resource.
|
||||
pub fn is_windmill_managed(self) -> bool {
|
||||
matches!(self, Self::Instance | Self::ExternalInstance)
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a self-teaching error for an unresolved `datatable://<name>` reference.
|
||||
@@ -1472,13 +1484,14 @@ pub struct GoverningDatatable {
|
||||
}
|
||||
|
||||
impl GoverningDatatable {
|
||||
/// Backed by the Windmill instance's own Postgres, which is the only substrate data table
|
||||
/// roles apply to.
|
||||
pub fn is_instance(&self) -> bool {
|
||||
/// The Windmill-managed cluster whose data table roles this entry can use. `None` for a
|
||||
/// resource-backed one: roles are logins Windmill creates, and it creates none on a host a
|
||||
/// workspace admin chose.
|
||||
pub fn role_cluster(&self) -> Option<crate::datatable_roles::DatatableRoleCluster> {
|
||||
self.datatable
|
||||
.database
|
||||
.as_ref()
|
||||
.is_some_and(|d| d.resource_type == DataTableCatalogResourceType::Instance)
|
||||
.and_then(|d| crate::datatable_roles::DatatableRoleCluster::of(d.resource_type))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1523,7 +1536,8 @@ pub async fn resolve_governing_datatable(
|
||||
}
|
||||
|
||||
/// 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.
|
||||
/// database, on Windmill's cluster or the external one; the user's own resource for a BYO-postgres
|
||||
/// one.
|
||||
async fn resolve_datatable_connection_unchecked(
|
||||
db: &DB,
|
||||
governing: &GoverningDatatable,
|
||||
@@ -1534,7 +1548,16 @@ async fn resolve_datatable_connection_unchecked(
|
||||
.database
|
||||
.as_ref()
|
||||
.expect("a governing entry owns a database");
|
||||
if database.resource_type == DataTableCatalogResourceType::Instance {
|
||||
if database.resource_type == DataTableCatalogResourceType::ExternalInstance {
|
||||
let pg_creds = crate::external_instance_pg::external_instance_connection_unchecked(
|
||||
db,
|
||||
&database.resource_path,
|
||||
replication,
|
||||
)
|
||||
.await?;
|
||||
serde_json::to_value(&pg_creds)
|
||||
.map_err(|e| Error::internal_err(format!("Error serializing pg creds: {}", e)))
|
||||
} else if database.resource_type == DataTableCatalogResourceType::Instance {
|
||||
let mut pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
|
||||
pg_creds.dbname = database.resource_path.clone();
|
||||
if replication {
|
||||
@@ -2074,6 +2097,10 @@ pub enum DucklakeCatalogResourceType {
|
||||
Postgresql,
|
||||
Mysql,
|
||||
Instance,
|
||||
/// On the external instance cluster ([`crate::external_instance_pg`]). Enterprise Edition.
|
||||
#[serde(rename = "external_instance")]
|
||||
#[strum(serialize = "external_instance")]
|
||||
ExternalInstance,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize)]
|
||||
@@ -2601,7 +2628,16 @@ async fn ducklake_conn_data(
|
||||
let ducklake = serde_json::from_value::<Ducklake>(ducklake)?;
|
||||
|
||||
let catalog_resource =
|
||||
if ducklake.catalog.resource_type == DucklakeCatalogResourceType::Instance {
|
||||
if ducklake.catalog.resource_type == DucklakeCatalogResourceType::ExternalInstance {
|
||||
let pg_creds = crate::external_instance_pg::external_instance_connection_unchecked(
|
||||
db,
|
||||
&ducklake.catalog.resource_path,
|
||||
false,
|
||||
)
|
||||
.await?;
|
||||
serde_json::to_value(&pg_creds)
|
||||
.map_err(|e| Error::internal_err(format!("Error serializing pg creds: {}", e)))?
|
||||
} else if ducklake.catalog.resource_type == DucklakeCatalogResourceType::Instance {
|
||||
let mut pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
|
||||
pg_creds.dbname = ducklake.catalog.resource_path.clone();
|
||||
pg_creds.user = Some("custom_instance_user".to_string());
|
||||
@@ -2982,6 +3018,14 @@ async fn register_fork_ducklake_namespace(
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
let mut tx = db.begin().await?;
|
||||
// A row naming an external database counts as a use of it. Written under the lock a drop takes,
|
||||
// and only while the database is still registered, so a drop cannot slip in between the
|
||||
// settings this attach resolved and the row that protects the database.
|
||||
if let Some(dbname) = catalog.strip_prefix("external_instance:") {
|
||||
crate::external_instance_pg::ensure_external_instance_database_registered(&mut tx, dbname)
|
||||
.await?;
|
||||
}
|
||||
sqlx::query!(
|
||||
"INSERT INTO fork_ducklake_namespace
|
||||
(workspace_id, ducklake_name, metadata_schema, catalog, storage, storage_ref, data_path)
|
||||
@@ -2996,9 +3040,10 @@ async fn register_fork_ducklake_namespace(
|
||||
&storage_ref,
|
||||
data_path,
|
||||
)
|
||||
.execute(db)
|
||||
.execute(&mut *tx)
|
||||
.await
|
||||
.map_err(|e| Error::internal_err(format!("registering fork ducklake namespace: {e:#}")))?;
|
||||
tx.commit().await?;
|
||||
let mut locations = FORK_DUCKLAKE_REGISTERED
|
||||
.get(w_id)
|
||||
.filter(|(_, exp)| *exp > now)
|
||||
|
||||
@@ -2240,11 +2240,64 @@ fn parse_attach_db_resource<'a>(query: &'a str) -> Option<ParsedAttachDbResource
|
||||
None
|
||||
}
|
||||
|
||||
/// The verification a DuckDB postgres attach keeps, as its libpq `sslmode` and `sslrootcert`.
|
||||
///
|
||||
/// Attaches have always turned verify-ca and verify-full into `require`, which resources rely on.
|
||||
/// A connection that explicitly refuses invalid certificates — the external instance cluster's —
|
||||
/// keeps its mode instead: under `require` its shared password would go to whichever server
|
||||
/// answers. DuckDB's libpq takes one root file, so it gets the system bundle plus the configured
|
||||
/// certificate.
|
||||
fn pg_attach_verification(res: &PgDatabase) -> Result<Option<(&str, std::path::PathBuf)>> {
|
||||
let mode = match res.sslmode.as_deref() {
|
||||
Some(mode @ ("verify-ca" | "verify-full")) if res.accept_invalid_certs == Some(false) => mode,
|
||||
_ => return Ok(None),
|
||||
};
|
||||
let bundle = windmill_common::system_ca_bundle()
|
||||
.map(std::fs::read_to_string)
|
||||
.transpose()
|
||||
.map_err(|e| Error::ExecutionErr(format!("Failed to read the system CA bundle: {e}")))?
|
||||
.unwrap_or_default();
|
||||
let pem = res.root_certificate_pem.as_deref().unwrap_or_default();
|
||||
if bundle.is_empty() && pem.is_empty() {
|
||||
return Err(Error::ExecutionErr(format!(
|
||||
"sslmode {mode} needs a root certificate, and this worker has no system CA bundle"
|
||||
)));
|
||||
}
|
||||
let roots = format!("{bundle}\n{pem}\n");
|
||||
use sha2::Digest;
|
||||
let path = std::env::temp_dir().join(format!(
|
||||
"windmill-pg-roots-{}.pem",
|
||||
hex::encode(&sha2::Sha256::digest(roots.as_bytes())[..8])
|
||||
));
|
||||
if !path.is_file() {
|
||||
// Renamed into place: a job attaching concurrently must never read a half-written file.
|
||||
let partial = path.with_extension(format!("{}.partial", Uuid::new_v4()));
|
||||
std::fs::write(&partial, &roots)
|
||||
.and_then(|()| std::fs::rename(&partial, &path))
|
||||
.map_err(|e| Error::ExecutionErr(format!("Failed to write root certificates: {e}")))?;
|
||||
}
|
||||
Ok(Some((mode, path)))
|
||||
}
|
||||
|
||||
fn pg_attach_uri(res: &PgDatabase) -> Result<String> {
|
||||
let uri = res.to_uri();
|
||||
let Some((mode, roots)) = pg_attach_verification(res)? else {
|
||||
return Ok(uri);
|
||||
};
|
||||
let base = uri.strip_suffix("?sslmode=require").ok_or_else(|| {
|
||||
Error::internal_err("unexpected sslmode in a postgres connection URI".to_string())
|
||||
})?;
|
||||
Ok(format!(
|
||||
"{base}?sslmode={mode}&sslrootcert={}",
|
||||
urlencoding::encode(&roots.to_string_lossy())
|
||||
))
|
||||
}
|
||||
|
||||
fn format_attach_db_conn_str(db_resource: Value, db_type: &str) -> Result<String> {
|
||||
let s = match db_type.to_lowercase().as_str() {
|
||||
"postgres" | "postgresql" => {
|
||||
let res: PgDatabase = serde_json::from_value(db_resource)?;
|
||||
res.to_uri()
|
||||
pg_attach_uri(&res)?
|
||||
}
|
||||
#[cfg(feature = "mysql")]
|
||||
"mysql" => {
|
||||
@@ -2391,7 +2444,9 @@ async fn transform_attach_ducklake(
|
||||
format!(", {}", user_extra_args)
|
||||
};
|
||||
let db_type = match ducklake.catalog.resource_type {
|
||||
DucklakeCatalogResourceType::Instance => "postgres",
|
||||
DucklakeCatalogResourceType::Instance | DucklakeCatalogResourceType::ExternalInstance => {
|
||||
"postgres"
|
||||
}
|
||||
_ => ducklake.catalog.resource_type.as_ref(),
|
||||
};
|
||||
|
||||
@@ -2507,7 +2562,8 @@ fn fork_defer_statements(
|
||||
hidden_passwords.lock().unwrap().push(pwd.to_string());
|
||||
}
|
||||
let db_type = match a.catalog.resource_type {
|
||||
DucklakeCatalogResourceType::Instance => "postgres",
|
||||
DucklakeCatalogResourceType::Instance
|
||||
| DucklakeCatalogResourceType::ExternalInstance => "postgres",
|
||||
_ => a.catalog.resource_type.as_ref(),
|
||||
};
|
||||
stmts.push(get_attach_db_install_str(db_type)?.to_string());
|
||||
@@ -2700,10 +2756,21 @@ fn pg_secret_attach_statements(db_resource: Value, alias_name: &str) -> Result<V
|
||||
let esc = |s: &str| s.replace('\'', "''");
|
||||
// The postgres secret type has no sslmode parameter, so it goes in the ATTACH
|
||||
// string; only the libpq values PgDatabase::to_uri collapses to are forwarded.
|
||||
let sslmode = match res.sslmode.as_deref() {
|
||||
Some("disable") => "disable",
|
||||
Some("require") | Some("verify-ca") | Some("verify-full") => "require",
|
||||
_ => "prefer",
|
||||
let sslmode = match pg_attach_verification(&res)? {
|
||||
// A libpq keyword/value string: the path is quoted for libpq, then for the DuckDB literal.
|
||||
Some((mode, roots)) => format!(
|
||||
"{mode} sslrootcert=''{}''",
|
||||
roots
|
||||
.to_string_lossy()
|
||||
.replace('\\', "\\\\")
|
||||
.replace('\'', "\\''")
|
||||
),
|
||||
None => match res.sslmode.as_deref() {
|
||||
Some("disable") => "disable",
|
||||
Some("require") | Some("verify-ca") | Some("verify-full") => "require",
|
||||
_ => "prefer",
|
||||
}
|
||||
.to_string(),
|
||||
};
|
||||
let secret_name = datatable_secret_name(alias_name);
|
||||
Ok(vec![
|
||||
@@ -2791,6 +2858,36 @@ pub struct Arg {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn pg_attach_keeps_verification_only_when_required() {
|
||||
let pg = |sslmode: &str, accept_invalid_certs: Option<bool>| PgDatabase {
|
||||
host: "db.internal".to_string(),
|
||||
user: Some("custom_instance_user".to_string()),
|
||||
password: Some("pw".to_string()),
|
||||
port: None,
|
||||
sslmode: Some(sslmode.to_string()),
|
||||
dbname: "dt".to_string(),
|
||||
root_certificate_pem: Some("-----BEGIN CERTIFICATE-----test".to_string()),
|
||||
accept_invalid_certs,
|
||||
use_iam_auth: None,
|
||||
region: None,
|
||||
};
|
||||
let uri = pg_attach_uri(&pg("verify-full", Some(false))).unwrap();
|
||||
assert!(uri.contains("?sslmode=verify-full&sslrootcert="), "{uri}");
|
||||
let root = urlencoding::decode(uri.split("sslrootcert=").nth(1).unwrap()).unwrap();
|
||||
let roots = std::fs::read_to_string(root.as_ref()).unwrap();
|
||||
assert!(roots.contains("-----BEGIN CERTIFICATE-----test"));
|
||||
let external = serde_json::to_value(pg("verify-full", Some(false))).unwrap();
|
||||
let attach = &pg_secret_attach_statements(external, "dt").unwrap()[3];
|
||||
assert!(
|
||||
attach.starts_with(&format!("ATTACH 'sslmode=verify-full sslrootcert=''{}''", root)),
|
||||
"{attach}"
|
||||
);
|
||||
// A resource that never opted in keeps the historical downgrade.
|
||||
assert!(pg_attach_uri(&pg("verify-full", None)).unwrap().ends_with("?sslmode=require"));
|
||||
assert!(pg_attach_uri(&pg("require", Some(false))).unwrap().ends_with("?sslmode=require"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attach_datatable_parses_name_and_role() {
|
||||
let reference_of = |q: &str| parse_attach_datatable(q).unwrap().reference;
|
||||
|
||||
@@ -111,6 +111,8 @@ const migrateCommand = new Command()
|
||||
)
|
||||
.action(migrateDown as any);
|
||||
|
||||
type DataTableResourceType = "postgresql" | "instance" | "external_instance";
|
||||
|
||||
async function create(
|
||||
opts: GlobalOptions & { resource?: string; force?: boolean },
|
||||
name?: string,
|
||||
@@ -139,12 +141,12 @@ async function create(
|
||||
|
||||
const datatables: Record<
|
||||
string,
|
||||
{ database: { resource_type: "postgresql" | "instance"; resource_path?: string } }
|
||||
{ database: { resource_type: DataTableResourceType; resource_path?: string } }
|
||||
> = {};
|
||||
for (const d of existing) {
|
||||
datatables[d.name] = {
|
||||
database: {
|
||||
resource_type: d.resource_type as "postgresql" | "instance",
|
||||
resource_type: d.resource_type as DataTableResourceType,
|
||||
resource_path: d.resource_path ?? undefined,
|
||||
},
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user