From 23bc15e2580fbc7b125008058cd102249b7e101c Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 21 Jul 2026 11:40:19 +0000 Subject: [PATCH 01/12] Did an refactoring to enable snapshot for fetch_expanded_live_specs, I added the snapshot as a parameter and left the user_id as a parameter because I wasn't sure if we would need the user_id as part of the interface and I didn't want to remove it without consulting someone about it. --- ...d417ff92af5cb2f6c11a35944288b14fde343.json | 141 ------------------ .../control-plane-api/src/evolutions/mod.rs | 2 +- crates/control-plane-api/src/live_specs/db.rs | 20 ++- .../control-plane-api/src/live_specs/mod.rs | 5 +- .../src/publications/initialize.rs | 23 ++- .../control-plane-api/src/publications/mod.rs | 2 +- 6 files changed, 37 insertions(+), 156 deletions(-) delete mode 100644 .sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json diff --git a/.sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json b/.sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json deleted file mode 100644 index c1ba203be75..00000000000 --- a/.sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json +++ /dev/null @@ -1,141 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n with collections(id) as (\n select ls.id\n from unnest($2::text[]) as names(catalog_name)\n join live_specs ls on ls.catalog_name = names.catalog_name\n ),\n exp(id) as (\n select lsf.source_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.target_id\n union\n select lsf.target_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.source_id\n )\n select\n ls.id as \"id: Id\",\n ls.last_pub_id as \"last_pub_id: Id\",\n ls.last_build_id as \"last_build_id: Id\",\n ls.data_plane_id as \"data_plane_id: Id\",\n ls.catalog_name,\n ls.spec_type as \"spec_type?: CatalogType\",\n ls.spec as \"spec: TextJson>\",\n ls.built_spec as \"built_spec: TextJson>\",\n ls.inferred_schema_md5,\n (\n select max(capability) from internal.user_roles($1) r\n where starts_with(ls.catalog_name, r.role_prefix)\n ) as \"user_capability: Capability\",\n coalesce(\n (select json_agg(row_to_json(role_grants))\n from role_grants\n where starts_with(ls.catalog_name, subject_role)),\n '[]'\n ) as \"spec_capabilities!: Json>\",\n ls.dependency_hash\n from exp\n join live_specs ls on ls.id = exp.id\n where ls.spec is not null and not ls.catalog_name = any($3);\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 1, - "name": "last_pub_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 2, - "name": "last_build_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 3, - "name": "data_plane_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 4, - "name": "catalog_name", - "type_info": "Text" - }, - { - "ordinal": 5, - "name": "spec_type?: CatalogType", - "type_info": { - "Custom": { - "name": "catalog_spec_type", - "kind": { - "Enum": [ - "capture", - "collection", - "materialization", - "test" - ] - } - } - } - }, - { - "ordinal": 6, - "name": "spec: TextJson>", - "type_info": "Json" - }, - { - "ordinal": 7, - "name": "built_spec: TextJson>", - "type_info": "Json" - }, - { - "ordinal": 8, - "name": "inferred_schema_md5", - "type_info": "Text" - }, - { - "ordinal": 9, - "name": "user_capability: Capability", - "type_info": { - "Custom": { - "name": "grant_capability", - "kind": { - "Enum": [ - "none", - "x_01", - "x_02", - "x_03", - "x_04", - "x_05", - "x_06", - "x_07", - "x_08", - "x_09", - "read", - "x_11", - "x_12", - "x_13", - "x_14", - "x_15", - "x_16", - "x_17", - "x_18", - "x_19", - "write", - "x_21", - "x_22", - "x_23", - "x_24", - "x_25", - "x_26", - "x_27", - "x_28", - "x_29", - "admin" - ] - } - } - } - }, - { - "ordinal": 10, - "name": "spec_capabilities!: Json>", - "type_info": "Json" - }, - { - "ordinal": 11, - "name": "dependency_hash", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Uuid", - "TextArray", - "TextArray" - ] - }, - "nullable": [ - false, - false, - false, - false, - false, - true, - true, - true, - true, - null, - null, - true - ] - }, - "hash": "6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343" -} diff --git a/crates/control-plane-api/src/evolutions/mod.rs b/crates/control-plane-api/src/evolutions/mod.rs index 0c3b9f7ba66..262b0579a2b 100644 --- a/crates/control-plane-api/src/evolutions/mod.rs +++ b/crates/control-plane-api/src/evolutions/mod.rs @@ -187,11 +187,11 @@ pub async fn evolve( .collect::>(); let exclude_names = draft.all_spec_names().collect::>(); let expanded_live = crate::live_specs::get_connected_live_specs( - user_id, &collection_names, &exclude_names, capability_filter, db, + &prefixes_and_capabilities, ) .await?; draft.add_live(expanded_live); diff --git a/crates/control-plane-api/src/live_specs/db.rs b/crates/control-plane-api/src/live_specs/db.rs index 50749dfef6b..752240156f7 100644 --- a/crates/control-plane-api/src/live_specs/db.rs +++ b/crates/control-plane-api/src/live_specs/db.rs @@ -1,7 +1,7 @@ use crate::{TextJson, snapshot::PrefixesAndCapabilities}; use models::{Capability, CatalogType, Id}; use serde_json::value::RawValue; -use sqlx::types::{Json, Uuid}; +use sqlx::types::Json; use tables::RoleGrant; /// Deletes the given live spec row, along with the corresponding `controller_jobs` row. @@ -127,17 +127,21 @@ pub async fn fetch_inferred_schemas( /// Queries for all non-deleted `live_specs` that are connected to the given `collection_names` via /// `live_spec_flows`. pub async fn fetch_expanded_live_specs( - user_id: Uuid, collection_names: &[&str], exclude_names: &[&str], db: impl sqlx::Executor<'_, Database = sqlx::Postgres>, + permissions_set: &PrefixesAndCapabilities<'_>, ) -> sqlx::Result> { + let (prefixes, capabilities): (Vec, Vec) = permissions_set + .iter() + .map(|(prefix, capabilities)| (prefix.to_string(), capabilities.1)) + .unzip(); sqlx::query_as!( LiveSpec, r#" with collections(id) as ( select ls.id - from unnest($2::text[]) as names(catalog_name) + from unnest($1::text[]) as names(catalog_name) join live_specs ls on ls.catalog_name = names.catalog_name ), exp(id) as ( @@ -148,6 +152,9 @@ pub async fn fetch_expanded_live_specs( select lsf.target_id as id from collections c join live_spec_flows lsf on c.id = lsf.source_id + ), + user_roles as materialized ( + select role_prefix, capability from UNNEST($3::text[], $4::grant_capability[]) as t(role_prefix, capability) ) select ls.id as "id: Id", @@ -160,7 +167,7 @@ pub async fn fetch_expanded_live_specs( ls.built_spec as "built_spec: TextJson>", ls.inferred_schema_md5, ( - select max(capability) from internal.user_roles($1) r + select max(capability) from user_roles r where starts_with(ls.catalog_name, r.role_prefix) ) as "user_capability: Capability", coalesce( @@ -172,11 +179,12 @@ pub async fn fetch_expanded_live_specs( ls.dependency_hash from exp join live_specs ls on ls.id = exp.id - where ls.spec is not null and not ls.catalog_name = any($3); + where ls.spec is not null and not ls.catalog_name = any($2); "#, - user_id, collection_names as &[&str], exclude_names as &[&str], + &prefixes, + &capabilities as &[Capability], ) .fetch_all(db) .await diff --git a/crates/control-plane-api/src/live_specs/mod.rs b/crates/control-plane-api/src/live_specs/mod.rs index bbb618e6f52..9c7b1852f5d 100644 --- a/crates/control-plane-api/src/live_specs/mod.rs +++ b/crates/control-plane-api/src/live_specs/mod.rs @@ -3,7 +3,6 @@ mod db; use anyhow::Context; use models::Capability; use std::ops::Deref; -use uuid::Uuid; pub use db::{ InferredSchemaRow, LiveSpec, fetch_expanded_live_specs, fetch_inferred_schemas, @@ -77,14 +76,14 @@ pub async fn get_live_specs( } pub async fn get_connected_live_specs( - user_id: Uuid, collection_names: &[&str], exclude_names: &[&str], filter_capability: Option, db: &sqlx::PgPool, + permissions_set: &PrefixesAndCapabilities<'_>, ) -> anyhow::Result { let expanded_rows = - db::fetch_expanded_live_specs(user_id, collection_names, exclude_names, db).await?; + db::fetch_expanded_live_specs(collection_names, exclude_names, db, permissions_set).await?; let mut live = tables::LiveCatalog::default(); for exp in expanded_rows { if let Some(minimum_capability) = filter_capability { diff --git a/crates/control-plane-api/src/publications/initialize.rs b/crates/control-plane-api/src/publications/initialize.rs index dd142d50dac..ef554522c7d 100644 --- a/crates/control-plane-api/src/publications/initialize.rs +++ b/crates/control-plane-api/src/publications/initialize.rs @@ -1,7 +1,8 @@ +use crate::Snapshot; use anyhow::Context; use itertools::Itertools; use models::Capability; -use std::future::Future; +use std::{future::Future, sync::Arc}; use uuid::Uuid; /// Initialize a draft prior to build/validation. This may add additional specs to the draft. @@ -11,6 +12,7 @@ pub trait Initialize: Send + Sync { db: &sqlx::PgPool, user_id: Uuid, draft: &mut tables::DraftCatalog, + snapshot_watch: &Arc>, ) -> impl Future> + Send; } @@ -22,6 +24,7 @@ impl Initialize for NoopInitialize { _db: &sqlx::PgPool, _user_id: Uuid, _draft: &mut tables::DraftCatalog, + _snapshot_watch: &Arc>, ) -> anyhow::Result<()> { Ok(()) } @@ -37,9 +40,14 @@ where db: &sqlx::PgPool, user_id: Uuid, draft: &mut tables::DraftCatalog, + snapshot_watch: &Arc>, ) -> anyhow::Result<()> { - self.0.initialize(db, user_id, draft).await?; - self.1.initialize(db, user_id, draft).await?; + self.0 + .initialize(db, user_id, draft, snapshot_watch) + .await?; + self.1 + .initialize(db, user_id, draft, snapshot_watch) + .await?; Ok(()) } } @@ -65,6 +73,7 @@ impl Initialize for ExpandDraft { db: &sqlx::PgPool, user_id: Uuid, draft: &mut tables::DraftCatalog, + snapshot_watch: &Arc>, ) -> anyhow::Result<()> { // Expand the set of drafted specs to include any tasks that read from or write to any of // the published collections. We do this so that validation can catch any inconsistencies @@ -81,12 +90,17 @@ impl Initialize for ExpandDraft { } else { None }; + let snapshot = snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); + let prefixes_and_capabilities = snapshot.prefix_and_capabilities_per_user(user_id); + let expanded_catalog = crate::live_specs::get_connected_live_specs( - user_id, + // user_id, &drafted_collections, &all_drafted_specs, capability_filter, db, + &prefixes_and_capabilities, ) .await?; tracing::debug!( @@ -112,6 +126,7 @@ impl Initialize for RuntimeV2Rollout { db: &sqlx::PgPool, _user_id: Uuid, draft: &mut tables::DraftCatalog, + _snapshot_watch: &Arc>, ) -> anyhow::Result<()> { if !self.new_captures { return Ok(()); diff --git a/crates/control-plane-api/src/publications/mod.rs b/crates/control-plane-api/src/publications/mod.rs index 6ffeee6b055..a0c968416f7 100644 --- a/crates/control-plane-api/src/publications/mod.rs +++ b/crates/control-plane-api/src/publications/mod.rs @@ -317,7 +317,7 @@ impl Publisher { ) -> anyhow::Result { let mut draft = raw_draft.clone_specs(); initialize - .initialize(&self.db, *user_id, &mut draft) + .initialize(&self.db, *user_id, &mut draft, &self.snapshot) .await .context("initializing draft")?; // It's important that we generate the pub id inside the retry loop so that we can From cec5af4c690287cc2dfd7d85aa6d4a19be3da1ef Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 21 Jul 2026 12:10:19 +0000 Subject: [PATCH 02/12] Did a refactoring for the function to enable use of snapshot generated permissions. --- ...ccf86b5d2d08d19999930aabe4b7373d8c671.json | 72 ------------------- crates/control-plane-api/src/evolutions/db.rs | 26 ++++--- 2 files changed, 18 insertions(+), 80 deletions(-) delete mode 100644 .sqlx/query-bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671.json diff --git a/.sqlx/query-bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671.json b/.sqlx/query-bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671.json deleted file mode 100644 index 461e13dbfe5..00000000000 --- a/.sqlx/query-bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671.json +++ /dev/null @@ -1,72 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n with drafted as (\n select\n ds.catalog_name,\n ds.id as draft_spec_id,\n ls.id as live_spec_id,\n ds.expect_pub_id,\n ls.last_pub_id as last_pub_id,\n ds.spec as spec,\n ds.spec_type as spec_type\n from draft_specs ds\n left join live_specs ls\n on ds.catalog_name = ls.catalog_name\n -- filter out live_specs rows that the user does not have admin access to\n and exists (select 1 from internal.user_roles($2, 'admin') r where ls.catalog_name ^@ r.role_prefix)\n where ds.draft_id = $1\n ),\n not_drafted as (\n select catalog_name from unnest($3::text[]) as names(catalog_name)\n except\n select catalog_name from drafted\n ),\n live as (\n select\n ls.catalog_name,\n ls.spec,\n ls.spec_type,\n ls.last_pub_id,\n ls.id\n from not_drafted\n join live_specs ls on not_drafted.catalog_name = ls.catalog_name\n where\n -- filter out live_specs rows that the user does not have admin access to\n exists (select 1 from internal.user_roles($2, 'admin') r where ls.catalog_name ^@ r.role_prefix)\n )\n select\n catalog_name as \"catalog_name!: String\",\n draft_spec_id as \"draft_spec_id: Id\",\n live_spec_id as \"live_spec_id: Id\",\n expect_pub_id as \"expect_pub_id: Id\",\n last_pub_id as \"last_pub_id: Id\",\n spec as \"spec: Json>\",\n spec_type as \"spec_type: CatalogType\"\n from drafted\n union all\n select\n catalog_name as \"catalog_name!: String\",\n null as \"draft_spec_id: Id\",\n id as \"live_spec_id: Id\",\n null as \"expect_pub_id: Id\",\n last_pub_id as \"last_pub_id: Id\",\n spec as \"spec: Json>\",\n spec_type as \"spec_type: CatalogType\"\n from live\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "catalog_name!: String", - "type_info": "Text" - }, - { - "ordinal": 1, - "name": "draft_spec_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 2, - "name": "live_spec_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 3, - "name": "expect_pub_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 4, - "name": "last_pub_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 5, - "name": "spec: Json>", - "type_info": "Json" - }, - { - "ordinal": 6, - "name": "spec_type: CatalogType", - "type_info": { - "Custom": { - "name": "catalog_spec_type", - "kind": { - "Enum": [ - "capture", - "collection", - "materialization", - "test" - ] - } - } - } - } - ], - "parameters": { - "Left": [ - "Macaddr8", - "Uuid", - "TextArray" - ] - }, - "nullable": [ - null, - null, - null, - null, - null, - null, - null - ] - }, - "hash": "bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671" -} diff --git a/crates/control-plane-api/src/evolutions/db.rs b/crates/control-plane-api/src/evolutions/db.rs index 5a641d6f5b9..833d4b7139e 100644 --- a/crates/control-plane-api/src/evolutions/db.rs +++ b/crates/control-plane-api/src/evolutions/db.rs @@ -1,6 +1,6 @@ -use crate::TextJson as Json; +use crate::{TextJson as Json, snapshot::PrefixesAndCapabilities}; use chrono::{DateTime, Utc}; -use models::{CatalogType, Id}; +use models::{Capability, CatalogType, Id}; use serde::Serialize; use serde_json::value::RawValue; use sqlx::types::Uuid; @@ -79,15 +79,24 @@ pub struct SpecRow { /// bindings. Technically, we could implement that filtering as part of the sql /// query, but the extra complexity doesn't seem warranted at this time. pub async fn resolve_specs( - user_id: Uuid, draft_id: Id, collection_names: Vec, + permissions_set: &PrefixesAndCapabilities<'_>, txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, ) -> sqlx::Result> { + // Doing the filter before handing this off to the query to reduce the amount of data being sent to the db. + let (prefixes, capabilities): (Vec, Vec) = permissions_set + .iter() + .filter(|(_prefix, capabilities)| capabilities.1 >= Capability::Admin) + .map(|(prefix, capabilities)| (prefix.to_string(), capabilities.1)) + .unzip(); sqlx::query_as!( SpecRow, r#" - with drafted as ( + with user_roles as materialized ( + select role_prefix, capability from UNNEST($3::text[], $4::grant_capability[]) as t(role_prefix, capability) + ), + drafted as ( select ds.catalog_name, ds.id as draft_spec_id, @@ -100,11 +109,11 @@ pub async fn resolve_specs( left join live_specs ls on ds.catalog_name = ls.catalog_name -- filter out live_specs rows that the user does not have admin access to - and exists (select 1 from internal.user_roles($2, 'admin') r where ls.catalog_name ^@ r.role_prefix) + and exists (select 1 from user_roles r where ls.catalog_name ^@ r.role_prefix) where ds.draft_id = $1 ), not_drafted as ( - select catalog_name from unnest($3::text[]) as names(catalog_name) + select catalog_name from unnest($2::text[]) as names(catalog_name) except select catalog_name from drafted ), @@ -119,7 +128,7 @@ pub async fn resolve_specs( join live_specs ls on not_drafted.catalog_name = ls.catalog_name where -- filter out live_specs rows that the user does not have admin access to - exists (select 1 from internal.user_roles($2, 'admin') r where ls.catalog_name ^@ r.role_prefix) + exists (select 1 from user_roles r where ls.catalog_name ^@ r.role_prefix) ) select catalog_name as "catalog_name!: String", @@ -142,8 +151,9 @@ pub async fn resolve_specs( from live "#, draft_id as Id, - user_id as Uuid, collection_names as Vec, + &prefixes, + &capabilities as &[Capability], ).fetch_all(&mut **txn) .await } From c90678d4974eb9e9b91c1edfd104c1a22c595513 Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 21 Jul 2026 12:57:51 +0000 Subject: [PATCH 03/12] Did the refactoring to use the snapshot instead of the DB call added tests verifing correct behavior. --- ...c480c5ae585051ce828cae19ea96e25996aaa.json | 23 --- crates/agent/src/directives/mod.rs | 14 +- .../agent/src/directives/storage_mappings.rs | 162 +++++++++++++++++- crates/agent/src/integration_tests/harness.rs | 7 +- crates/agent/src/main.rs | 3 +- .../src/directives/storage_mappings.rs | 16 -- 6 files changed, 174 insertions(+), 51 deletions(-) delete mode 100644 .sqlx/query-d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa.json diff --git a/.sqlx/query-d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa.json b/.sqlx/query-d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa.json deleted file mode 100644 index ad62491448e..00000000000 --- a/.sqlx/query-d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "select true as whatever_column from internal.user_roles($1, 'admin') where starts_with(role_prefix, $2)", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "whatever_column", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa" -} diff --git a/crates/agent/src/directives/mod.rs b/crates/agent/src/directives/mod.rs index 40ff93ffa2f..1120a3d86cc 100644 --- a/crates/agent/src/directives/mod.rs +++ b/crates/agent/src/directives/mod.rs @@ -1,3 +1,5 @@ +use std::sync::Arc; + use control_plane_api::{ directives::{Row, fetch_directive, resolve}, logs, @@ -49,13 +51,21 @@ pub enum Directive { pub struct DirectiveHandler { accounts_user_email: String, logs_tx: logs::Tx, + /// Live authorization Snapshot watch, retained so tests can force it to + /// re-fetch from Postgres after mutating grants. See `refresh_snapshot`. + snapshot_watch: Arc>, } impl DirectiveHandler { - pub fn new(accounts_user_email: String, logs_tx: &logs::Tx) -> Self { + pub fn new( + accounts_user_email: String, + logs_tx: &logs::Tx, + snapshot_watch: Arc>, + ) -> Self { Self { accounts_user_email, logs_tx: logs_tx.clone(), + snapshot_watch, } } } @@ -133,7 +143,7 @@ impl DirectiveHandler { Ok(Directive::ClickToAccept(d)) => click_to_accept::apply(d, row, txn).await?, Ok(Directive::AcceptDemoTenant(d)) => accept_demo_tenant::apply(d, row, txn).await?, Ok(Directive::StorageMappings(d)) => { - storage_mappings::apply(d, row, &self.logs_tx, txn).await? + storage_mappings::apply(d, row, &self.logs_tx, txn, &self.snapshot_watch).await? } }; Ok(status) diff --git a/crates/agent/src/directives/storage_mappings.rs b/crates/agent/src/directives/storage_mappings.rs index aece82ea866..32d52479c0f 100644 --- a/crates/agent/src/directives/storage_mappings.rs +++ b/crates/agent/src/directives/storage_mappings.rs @@ -1,14 +1,11 @@ -use std::io::Write; +use std::{io::Write, sync::Arc}; use crate::directives::JobStatus; use anyhow::Context; use control_plane_api::{ directives::{ Row, - storage_mappings::{ - StorageMapping, fetch_storage_mappings, upsert_storage_mapping, - user_has_admin_capability, - }, + storage_mappings::{StorageMapping, fetch_storage_mappings, upsert_storage_mapping}, }, jobs, logs, }; @@ -33,12 +30,13 @@ pub async fn apply( row: Row, logs_tx: &logs::Tx, txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, + snapshot_watch: &Arc>, ) -> anyhow::Result { let detail = format!( "updated by user {} via applied directive {}", row.user_id, row.apply_id ); - let (collection_data, recovery) = match validate(txn, logs_tx, row).await { + let (collection_data, recovery) = match validate(txn, logs_tx, row, &snapshot_watch).await { Ok(c) => c, Err(err) => { return Ok(JobStatus::invalid_claims(err)); @@ -75,6 +73,7 @@ async fn validate( txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, logs_tx: &logs::Tx, row: Row, + snapshot_watch: &Arc>, ) -> anyhow::Result<(ProposedMapping, ProposedMapping)> { let claims: Claims = serde_json::from_str(row.user_claims.get()).context("parsing user_claims")?; @@ -94,8 +93,11 @@ async fn validate( // Note: we must assert that user has admin capability for the _entire tenant_, even if in the // future we allow for updating mappings of narrower prefixes. This is required because a new // storage mapping for `a/b/` may implicitly override the existing mapping for `a/`. + let refresh = snapshot_watch.token(); + let snapshot = refresh.result().unwrap(); + let user_has_admin = - user_has_admin_capability(row.user_id, &claims.catalog_prefix, txn).await?; + user_can_admin_tenant(&snapshot, row.user_id, claims.catalog_prefix.as_str()); anyhow::ensure!( user_has_admin, "user does not have required 'admin' capability to '{}'", @@ -128,6 +130,28 @@ async fn validate( Ok((collection_data, recovery_data)) } +/// Returns whether `user_id` holds `admin` capability on `catalog_prefix` +/// itself, or on any prefix nested beneath it, according to `snapshot`. +/// +/// This deliberately mirrors the legacy +/// `internal.user_roles($user, 'admin') where starts_with(role_prefix, catalog_prefix)` +/// SQL check that it replaced: a user who administers only a *sub-prefix* of +/// the tenant (e.g. `acmeCo/team/` when altering `acmeCo/`) remains authorized. +/// Note this is the reverse of `tables::UserGrant::is_authorized`, which +/// considers only grants at or *above* `catalog_prefix`. +fn user_can_admin_tenant( + snapshot: &control_plane_api::Snapshot, + user_id: Uuid, + catalog_prefix: &str, +) -> bool { + snapshot + .prefix_and_capabilities_per_user(user_id) + .iter() + .any(|(role_prefix, capabilities)| { + role_prefix.starts_with(catalog_prefix) && capabilities.1 >= models::Capability::Admin + }) +} + // This is a macro instead of a function to work around the fact that file paths are `OsStr`s // instead of regular `&str`s. macro_rules! check_command { @@ -323,3 +347,127 @@ This file is written to your storage bucket in order to test that we have the ne permissions to create and delete objects. If you're seeing this file stick around, then it's likely because we lacked the necessary permissions to delete it. You may remove this file at any time, and doing so will not impact the function of Estuary."#; + +#[cfg(test)] +mod test { + use super::*; + use control_plane_api::snapshot::SnapshotData; + + // Tenant prefix under test. Storage mappings may only be altered for a + // top-level tenant, so this is always a single-segment prefix. + const TENANT: &str = "acmeCo/"; + + fn user_grant( + user_id: Uuid, + object_role: &str, + capability: models::Capability, + ) -> tables::UserGrant { + tables::UserGrant { + user_id, + object_role: models::Prefix::new(object_role), + capability, + bundles: Vec::new(), + } + } + + fn role_grant( + subject_role: &str, + object_role: &str, + capability: models::Capability, + ) -> tables::RoleGrant { + tables::RoleGrant { + subject_role: models::Prefix::new(subject_role), + object_role: models::Prefix::new(object_role), + capability, + bundles: Vec::new(), + } + } + + fn snapshot( + user_grants: Vec, + role_grants: Vec, + ) -> control_plane_api::Snapshot { + let data = SnapshotData { + collections: Vec::new(), + data_planes: Vec::new(), + migrations: Vec::new(), + role_grants, + user_grants, + tasks: Vec::new(), + }; + control_plane_api::Snapshot::new(chrono::DateTime::UNIX_EPOCH, data) + } + + // Case 1: a user granted `admin` directly on the tenant root is authorized. + // This is the case where the old SQL check and `is_authorized` agreed. + #[test] + fn admin_on_tenant_root_is_authorized() { + let user_id = Uuid::from_bytes([1; 16]); + let snapshot = snapshot( + vec![user_grant(user_id, TENANT, models::Capability::Admin)], + Vec::new(), + ); + assert!(user_can_admin_tenant(&snapshot, user_id, TENANT)); + } + + // Case 2: a user granted `admin` only on a prefix *beneath* the tenant root + // is still authorized. This is the behavior that would have been lost with a + // plain `is_authorized` check, and which the sub-prefix scan preserves. + #[test] + fn admin_on_sub_prefix_is_authorized() { + let user_id = Uuid::from_bytes([2; 16]); + let snapshot = snapshot( + vec![user_grant( + user_id, + "acmeCo/team/", + models::Capability::Admin, + )], + Vec::new(), + ); + assert!(user_can_admin_tenant(&snapshot, user_id, TENANT)); + } + + // Case 3: a user reaches `admin` on the tenant transitively, through a role + // grant (e.g. an `estuary_support/` role that is itself granted admin over + // the tenant). The grant-graph projection must be honored. + #[test] + fn admin_via_role_grant_projection_is_authorized() { + let user_id = Uuid::from_bytes([3; 16]); + let snapshot = snapshot( + vec![user_grant( + user_id, + "estuary_support/", + models::Capability::Admin, + )], + vec![role_grant( + "estuary_support/", + TENANT, + models::Capability::Admin, + )], + ); + assert!(user_can_admin_tenant(&snapshot, user_id, TENANT)); + } + + // Negative control: a capability below `admin` on the tenant is not enough. + #[test] + fn write_capability_is_denied() { + let user_id = Uuid::from_bytes([4; 16]); + let snapshot = snapshot( + vec![user_grant(user_id, TENANT, models::Capability::Write)], + Vec::new(), + ); + assert!(!user_can_admin_tenant(&snapshot, user_id, TENANT)); + } + + // Negative control: admin over a sibling tenant confers no authority here, + // proving the sub-prefix scan does not match unrelated or ancestor prefixes. + #[test] + fn admin_on_unrelated_tenant_is_denied() { + let user_id = Uuid::from_bytes([5; 16]); + let snapshot = snapshot( + vec![user_grant(user_id, "bobCo/", models::Capability::Admin)], + Vec::new(), + ); + assert!(!user_can_admin_tenant(&snapshot, user_id, TENANT)); + } +} diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index 704036df48c..c9641806cec 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -276,8 +276,11 @@ impl HarnessBuilder { let controller_exec = crate::controllers::executor::LiveSpecControllerExecutor::new(control_plane.clone()); - let directive_exec = - crate::directives::DirectiveHandler::new("support@estuary.test".to_string(), &logs_tx); + let directive_exec = crate::directives::DirectiveHandler::new( + "support@estuary.test".to_string(), + &logs_tx, + snapshot_watch.clone(), + ); let mut harness = TestHarness { test_name, diff --git a/crates/agent/src/main.rs b/crates/agent/src/main.rs index b7731758498..5c3fef8d8c0 100644 --- a/crates/agent/src/main.rs +++ b/crates/agent/src/main.rs @@ -395,7 +395,8 @@ async fn async_main(args: Args) -> Result<(), anyhow::Error> { let api_server = async move { anyhow::Result::Ok(api_server.await?) }; let automations_fut = if args.max_automations > 0 { - let directive_executor = agent::DirectiveHandler::new(args.accounts_email, &logs_tx); + let directive_executor = + agent::DirectiveHandler::new(args.accounts_email, &logs_tx, snapshot_watch.clone()); let connector_tags_executor = agent::TagExecutor::new(&args.connector_network, &logs_tx); let mut automations_server = automations::Server::new() .register(agent::controllers::LiveSpecControllerExecutor::new( diff --git a/crates/control-plane-api/src/directives/storage_mappings.rs b/crates/control-plane-api/src/directives/storage_mappings.rs index 74c1fa72f8f..7106e4ce21a 100644 --- a/crates/control-plane-api/src/directives/storage_mappings.rs +++ b/crates/control-plane-api/src/directives/storage_mappings.rs @@ -1,21 +1,5 @@ use crate::TextJson; use serde_json::value::RawValue; -use sqlx::types::Uuid; - -pub async fn user_has_admin_capability( - user_id: Uuid, - catalog_prefix: &str, - txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, -) -> sqlx::Result { - let row = sqlx::query!( - r#"select true as whatever_column from internal.user_roles($1, 'admin') where starts_with(role_prefix, $2)"#, - user_id, - catalog_prefix, - ) - .fetch_optional(&mut **txn) - .await?; - Ok(row.is_some()) -} pub async fn upsert_storage_mapping( detail: Option<&str>, From 5dd30409fe7768264a41c224fc54c580c872cdc4 Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 21 Jul 2026 13:06:40 +0000 Subject: [PATCH 04/12] Missed some files. --- ...41840a1743da8a5d1863156d6207853b19186.json | 119 +++++++++++ ...ca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json | 188 ++++++++++++++++++ 2 files changed, 307 insertions(+) create mode 100644 .sqlx/query-11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186.json create mode 100644 .sqlx/query-d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json diff --git a/.sqlx/query-11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186.json b/.sqlx/query-11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186.json new file mode 100644 index 00000000000..939eeec2dd0 --- /dev/null +++ b/.sqlx/query-11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186.json @@ -0,0 +1,119 @@ +{ + "db_name": "PostgreSQL", + "query": "\n with user_roles as materialized (\n select role_prefix, capability from UNNEST($3::text[], $4::grant_capability[]) as t(role_prefix, capability)\n ),\n drafted as (\n select\n ds.catalog_name,\n ds.id as draft_spec_id,\n ls.id as live_spec_id,\n ds.expect_pub_id,\n ls.last_pub_id as last_pub_id,\n ds.spec as spec,\n ds.spec_type as spec_type\n from draft_specs ds\n left join live_specs ls\n on ds.catalog_name = ls.catalog_name\n -- filter out live_specs rows that the user does not have admin access to\n and exists (select 1 from user_roles r where ls.catalog_name ^@ r.role_prefix)\n where ds.draft_id = $1\n ),\n not_drafted as (\n select catalog_name from unnest($2::text[]) as names(catalog_name)\n except\n select catalog_name from drafted\n ),\n live as (\n select\n ls.catalog_name,\n ls.spec,\n ls.spec_type,\n ls.last_pub_id,\n ls.id\n from not_drafted\n join live_specs ls on not_drafted.catalog_name = ls.catalog_name\n where\n -- filter out live_specs rows that the user does not have admin access to\n exists (select 1 from user_roles r where ls.catalog_name ^@ r.role_prefix)\n )\n select\n catalog_name as \"catalog_name!: String\",\n draft_spec_id as \"draft_spec_id: Id\",\n live_spec_id as \"live_spec_id: Id\",\n expect_pub_id as \"expect_pub_id: Id\",\n last_pub_id as \"last_pub_id: Id\",\n spec as \"spec: Json>\",\n spec_type as \"spec_type: CatalogType\"\n from drafted\n union all\n select\n catalog_name as \"catalog_name!: String\",\n null as \"draft_spec_id: Id\",\n id as \"live_spec_id: Id\",\n null as \"expect_pub_id: Id\",\n last_pub_id as \"last_pub_id: Id\",\n spec as \"spec: Json>\",\n spec_type as \"spec_type: CatalogType\"\n from live\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "catalog_name!: String", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "draft_spec_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 2, + "name": "live_spec_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 3, + "name": "expect_pub_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 4, + "name": "last_pub_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 5, + "name": "spec: Json>", + "type_info": "Json" + }, + { + "ordinal": 6, + "name": "spec_type: CatalogType", + "type_info": { + "Custom": { + "name": "catalog_spec_type", + "kind": { + "Enum": [ + "capture", + "collection", + "materialization", + "test" + ] + } + } + } + } + ], + "parameters": { + "Left": [ + "Macaddr8", + "TextArray", + "TextArray", + { + "Custom": { + "name": "grant_capability[]", + "kind": { + "Array": { + "Custom": { + "name": "grant_capability", + "kind": { + "Enum": [ + "none", + "x_01", + "x_02", + "x_03", + "x_04", + "x_05", + "x_06", + "x_07", + "x_08", + "x_09", + "read", + "x_11", + "x_12", + "x_13", + "x_14", + "x_15", + "x_16", + "x_17", + "x_18", + "x_19", + "write", + "x_21", + "x_22", + "x_23", + "x_24", + "x_25", + "x_26", + "x_27", + "x_28", + "x_29", + "admin" + ] + } + } + } + } + } + } + ] + }, + "nullable": [ + null, + null, + null, + null, + null, + null, + null + ] + }, + "hash": "11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186" +} diff --git a/.sqlx/query-d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json b/.sqlx/query-d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json new file mode 100644 index 00000000000..0fe92b4b2c0 --- /dev/null +++ b/.sqlx/query-d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json @@ -0,0 +1,188 @@ +{ + "db_name": "PostgreSQL", + "query": "\n with collections(id) as (\n select ls.id\n from unnest($1::text[]) as names(catalog_name)\n join live_specs ls on ls.catalog_name = names.catalog_name\n ),\n exp(id) as (\n select lsf.source_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.target_id\n union\n select lsf.target_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.source_id\n ),\n user_roles as materialized (\n select role_prefix, capability from UNNEST($3::text[], $4::grant_capability[]) as t(role_prefix, capability)\n )\n select\n ls.id as \"id: Id\",\n ls.last_pub_id as \"last_pub_id: Id\",\n ls.last_build_id as \"last_build_id: Id\",\n ls.data_plane_id as \"data_plane_id: Id\",\n ls.catalog_name,\n ls.spec_type as \"spec_type?: CatalogType\",\n ls.spec as \"spec: TextJson>\",\n ls.built_spec as \"built_spec: TextJson>\",\n ls.inferred_schema_md5,\n (\n select max(capability) from user_roles r\n where starts_with(ls.catalog_name, r.role_prefix)\n ) as \"user_capability: Capability\",\n coalesce(\n (select json_agg(row_to_json(role_grants))\n from role_grants\n where starts_with(ls.catalog_name, subject_role)),\n '[]'\n ) as \"spec_capabilities!: Json>\",\n ls.dependency_hash\n from exp\n join live_specs ls on ls.id = exp.id\n where ls.spec is not null and not ls.catalog_name = any($2);\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "last_pub_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 2, + "name": "last_build_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 3, + "name": "data_plane_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 4, + "name": "catalog_name", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "spec_type?: CatalogType", + "type_info": { + "Custom": { + "name": "catalog_spec_type", + "kind": { + "Enum": [ + "capture", + "collection", + "materialization", + "test" + ] + } + } + } + }, + { + "ordinal": 6, + "name": "spec: TextJson>", + "type_info": "Json" + }, + { + "ordinal": 7, + "name": "built_spec: TextJson>", + "type_info": "Json" + }, + { + "ordinal": 8, + "name": "inferred_schema_md5", + "type_info": "Text" + }, + { + "ordinal": 9, + "name": "user_capability: Capability", + "type_info": { + "Custom": { + "name": "grant_capability", + "kind": { + "Enum": [ + "none", + "x_01", + "x_02", + "x_03", + "x_04", + "x_05", + "x_06", + "x_07", + "x_08", + "x_09", + "read", + "x_11", + "x_12", + "x_13", + "x_14", + "x_15", + "x_16", + "x_17", + "x_18", + "x_19", + "write", + "x_21", + "x_22", + "x_23", + "x_24", + "x_25", + "x_26", + "x_27", + "x_28", + "x_29", + "admin" + ] + } + } + } + }, + { + "ordinal": 10, + "name": "spec_capabilities!: Json>", + "type_info": "Json" + }, + { + "ordinal": 11, + "name": "dependency_hash", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "TextArray", + "TextArray", + "TextArray", + { + "Custom": { + "name": "grant_capability[]", + "kind": { + "Array": { + "Custom": { + "name": "grant_capability", + "kind": { + "Enum": [ + "none", + "x_01", + "x_02", + "x_03", + "x_04", + "x_05", + "x_06", + "x_07", + "x_08", + "x_09", + "read", + "x_11", + "x_12", + "x_13", + "x_14", + "x_15", + "x_16", + "x_17", + "x_18", + "x_19", + "write", + "x_21", + "x_22", + "x_23", + "x_24", + "x_25", + "x_26", + "x_27", + "x_28", + "x_29", + "admin" + ] + } + } + } + } + } + } + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + true, + true, + true, + true, + null, + null, + true + ] + }, + "hash": "d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4" +} From 2c678d9ca07e2b3d6ed7900ce844e9120cb19349 Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 21 Jul 2026 14:26:58 +0000 Subject: [PATCH 05/12] Did a refactor based test for the switching from the over the create_data_plane function. --- ...9de98bc61be2843033a74a9bf0c5c9d44c410.json | 2 +- ...78286d1024dd706152c005983637982365277.json | 2 +- .../src/server/create_data_plane.rs | 142 +++++++++++++++++- 3 files changed, 137 insertions(+), 9 deletions(-) diff --git a/.sqlx/query-b10fda873a90630129d968ed7369de98bc61be2843033a74a9bf0c5c9d44c410.json b/.sqlx/query-b10fda873a90630129d968ed7369de98bc61be2843033a74a9bf0c5c9d44c410.json index 4828299c4ee..32826828169 100644 --- a/.sqlx/query-b10fda873a90630129d968ed7369de98bc61be2843033a74a9bf0c5c9d44c410.json +++ b/.sqlx/query-b10fda873a90630129d968ed7369de98bc61be2843033a74a9bf0c5c9d44c410.json @@ -95,7 +95,7 @@ false, true, false, - false, + true, false, false, true, diff --git a/.sqlx/query-e9c8d33f3b85a5964538fc00fa678286d1024dd706152c005983637982365277.json b/.sqlx/query-e9c8d33f3b85a5964538fc00fa678286d1024dd706152c005983637982365277.json index f76bd045e58..8f9e316830b 100644 --- a/.sqlx/query-e9c8d33f3b85a5964538fc00fa678286d1024dd706152c005983637982365277.json +++ b/.sqlx/query-e9c8d33f3b85a5964538fc00fa678286d1024dd706152c005983637982365277.json @@ -64,7 +64,7 @@ false, null, false, - false, + true, true, false, null, diff --git a/crates/control-plane-api/src/server/create_data_plane.rs b/crates/control-plane-api/src/server/create_data_plane.rs index 15ca92f9802..29506d418b9 100644 --- a/crates/control-plane-api/src/server/create_data_plane.rs +++ b/crates/control-plane-api/src/server/create_data_plane.rs @@ -65,13 +65,7 @@ pub async fn create_data_plane( ) -> Result, crate::server::error::ApiError> { let models::authorizations::ControlClaims { sub: user_id, .. } = env.claims()?; - if let None = sqlx::query!( - "select role_prefix from internal.user_roles($1, 'admin') where role_prefix = 'ops/'", - user_id, - ) - .fetch_optional(&env.pg_pool) - .await? - { + if !user_can_admin_ops(&app.snapshot_watch, *user_id) { return Err(tonic::Status::permission_denied( "authenticated user is not an admin of the 'ops/' tenant", ) @@ -317,6 +311,28 @@ pub async fn create_data_plane( Ok(axum::Json(Response {})) } +/// Returns whether `user_id` holds `admin` capability resolving to exactly the +/// `ops/` tenant, which is required to create a data-plane. +/// +/// This is evaluated against the authorization `Snapshot` (as was done for the +/// storage-mappings directive), replacing the prior +/// `internal.user_roles($user, 'admin') where role_prefix = 'ops/'` SQL check. +/// Because that check was an *exact* `ops/` match — a data-plane always lives +/// under `ops/` — this looks for `admin` at precisely the `ops/` prefix, and +/// deliberately does not accept a grant at an ancestor or a descendant of it. +fn user_can_admin_ops( + snapshot_watch: &std::sync::Arc>, + user_id: uuid::Uuid, +) -> bool { + let refresh = snapshot_watch.token(); + let snapshot = refresh.result().unwrap(); + + snapshot + .prefix_and_capabilities_per_user(user_id) + .get("ops/") + .is_some_and(|(_bits, legacy)| *legacy >= models::Capability::Admin) +} + impl Validate for Category { fn validate(&self) -> Result<(), validator::ValidationErrors> { if let Self::Manual(manual) = &self { @@ -326,3 +342,115 @@ impl Validate for Category { } } } + +#[cfg(test)] +mod test { + use super::user_can_admin_ops; + + // Inserts `user_id` along with the given `user_grants` + // (as `(object_role, capability)`) and `role_grants` (as + // `(subject_role, object_role, capability)`), builds an authorization + // Snapshot from that database state, then evaluates the `ops/` admin check + // for that user. Running through a real Snapshot built from real grants + // proves the snapshot-based check preserves the prior SQL behavior across + // every scenario below. + async fn eval( + pool: &sqlx::PgPool, + user_id: uuid::Uuid, + user_grants: &[(&str, &str)], + role_grants: &[(&str, &str, &str)], + ) -> bool { + sqlx::query("insert into auth.users (id, email) values ($1, $2)") + .bind(user_id) + .bind(format!("{user_id}@example.com")) + .execute(pool) + .await + .unwrap(); + + for (object_role, capability) in user_grants { + sqlx::query( + "insert into user_grants (user_id, object_role, capability) + values ($1, $2, $3::grant_capability)", + ) + .bind(user_id) + .bind(object_role) + .bind(capability) + .execute(pool) + .await + .unwrap(); + } + for (subject_role, object_role, capability) in role_grants { + sqlx::query( + "insert into role_grants (subject_role, object_role, capability) + values ($1, $2, $3::grant_capability)", + ) + .bind(subject_role) + .bind(object_role) + .bind(capability) + .execute(pool) + .await + .unwrap(); + } + + // `gate: false` so the first (and only) refresh serves the real + // snapshot built from the grants inserted above, rather than the empty + // snapshot used to exercise the server's retry flow. + let snapshot_watch = crate::test_server::snapshot(pool.clone(), false).await; + user_can_admin_ops(&snapshot_watch, user_id) + } + + // A user granted `admin` directly on `ops/` is authorized. + #[sqlx::test(migrations = "../../supabase/migrations")] + async fn admin_directly_on_ops_is_authorized(pool: sqlx::PgPool) { + let user_id = uuid::Uuid::from_bytes([1; 16]); + assert!(eval(&pool, user_id, &[("ops/", "admin")], &[]).await); + } + + // A user reaching `admin` on `ops/` transitively through a role grant + // (e.g. an `estuary_support/` role that is itself granted admin over `ops/`) + // is authorized. This exercises the recursive expansion in `user_roles`. + #[sqlx::test(migrations = "../../supabase/migrations")] + async fn admin_on_ops_via_role_grant_is_authorized(pool: sqlx::PgPool) { + let user_id = uuid::Uuid::from_bytes([2; 16]); + assert!( + eval( + &pool, + user_id, + &[("estuary_support/", "admin")], + &[("estuary_support/", "ops/", "admin")], + ) + .await + ); + } + + // Admin over a *sub-prefix* of `ops/` (e.g. `ops/dp/`) does NOT authorize: + // the check is an exact `role_prefix = 'ops/'` match, not a prefix match. + // This deliberately differs from the storage-mappings sub-prefix behavior, + // and pins it so a later snapshot refactor cannot silently loosen it. + #[sqlx::test(migrations = "../../supabase/migrations")] + async fn admin_on_sub_prefix_of_ops_is_denied(pool: sqlx::PgPool) { + let user_id = uuid::Uuid::from_bytes([3; 16]); + assert!(!eval(&pool, user_id, &[("ops/dp/", "admin")], &[]).await); + } + + // A capability below `admin` on `ops/` is not enough. + #[sqlx::test(migrations = "../../supabase/migrations")] + async fn write_on_ops_is_denied(pool: sqlx::PgPool) { + let user_id = uuid::Uuid::from_bytes([4; 16]); + assert!(!eval(&pool, user_id, &[("ops/", "write")], &[]).await); + } + + // Admin over an unrelated tenant confers no authority over `ops/`. + #[sqlx::test(migrations = "../../supabase/migrations")] + async fn admin_on_unrelated_tenant_is_denied(pool: sqlx::PgPool) { + let user_id = uuid::Uuid::from_bytes([5; 16]); + assert!(!eval(&pool, user_id, &[("aliceCo/", "admin")], &[]).await); + } + + // A user with no grants at all is denied. + #[sqlx::test(migrations = "../../supabase/migrations")] + async fn no_grants_is_denied(pool: sqlx::PgPool) { + let user_id = uuid::Uuid::from_bytes([6; 16]); + assert!(!eval(&pool, user_id, &[], &[]).await); + } +} From 9d5065f53989a153cfe89c81bab9aca26f7563a5 Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 21 Jul 2026 14:34:19 +0000 Subject: [PATCH 06/12] Did a similar refactoring to the update_l2_reporting. --- ...bf45b2ea36fd6fb8be6e988627b2bdf45e21d.json | 22 ------------------- .../src/server/create_data_plane.rs | 2 +- .../src/server/update_l2_reporting.rs | 8 +------ 3 files changed, 2 insertions(+), 30 deletions(-) delete mode 100644 .sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json diff --git a/.sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json b/.sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json deleted file mode 100644 index 9dcadd077f7..00000000000 --- a/.sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "select role_prefix from internal.user_roles($1, 'admin') where role_prefix = 'ops/'", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "role_prefix", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [ - null - ] - }, - "hash": "752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d" -} diff --git a/crates/control-plane-api/src/server/create_data_plane.rs b/crates/control-plane-api/src/server/create_data_plane.rs index 29506d418b9..243856f2b8b 100644 --- a/crates/control-plane-api/src/server/create_data_plane.rs +++ b/crates/control-plane-api/src/server/create_data_plane.rs @@ -320,7 +320,7 @@ pub async fn create_data_plane( /// Because that check was an *exact* `ops/` match — a data-plane always lives /// under `ops/` — this looks for `admin` at precisely the `ops/` prefix, and /// deliberately does not accept a grant at an ancestor or a descendant of it. -fn user_can_admin_ops( +pub(super) fn user_can_admin_ops( snapshot_watch: &std::sync::Arc>, user_id: uuid::Uuid, ) -> bool { diff --git a/crates/control-plane-api/src/server/update_l2_reporting.rs b/crates/control-plane-api/src/server/update_l2_reporting.rs index 9bf0e42d082..445560d2234 100644 --- a/crates/control-plane-api/src/server/update_l2_reporting.rs +++ b/crates/control-plane-api/src/server/update_l2_reporting.rs @@ -34,13 +34,7 @@ pub async fn update_l2_reporting( ) -> Result, crate::ApiError> { let crate::ControlClaims { sub: user_id, .. } = env.claims()?; - if let None = sqlx::query!( - "select role_prefix from internal.user_roles($1, 'admin') where role_prefix = 'ops/'", - user_id, - ) - .fetch_optional(&env.pg_pool) - .await? - { + if !super::create_data_plane::user_can_admin_ops(&app.snapshot_watch, *user_id) { return Err(tonic::Status::permission_denied( "authenticated user is not an admin of the 'ops/' tenant", ) From a653f3df883e10cd7732dba2435d1c01df1917fe Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 21 Jul 2026 15:13:22 +0000 Subject: [PATCH 07/12] Doing a refactor for deny and retry. --- crates/control-plane-api/src/discovers/mod.rs | 1 - .../control-plane-api/src/publications/mod.rs | 1 - .../src/server/create_data_plane.rs | 56 ++++++++++++------- .../src/server/update_l2_reporting.rs | 21 ++++--- 4 files changed, 48 insertions(+), 31 deletions(-) diff --git a/crates/control-plane-api/src/discovers/mod.rs b/crates/control-plane-api/src/discovers/mod.rs index 55e13742ec4..2d6f6b3627b 100644 --- a/crates/control-plane-api/src/discovers/mod.rs +++ b/crates/control-plane-api/src/discovers/mod.rs @@ -315,7 +315,6 @@ impl DiscoverHandler { .iter() .map(|b| b.target.to_string()) .collect::>(); - // user_id, let live = crate::live_specs::get_live_specs( &collection_names, diff --git a/crates/control-plane-api/src/publications/mod.rs b/crates/control-plane-api/src/publications/mod.rs index a0c968416f7..282d068d8e5 100644 --- a/crates/control-plane-api/src/publications/mod.rs +++ b/crates/control-plane-api/src/publications/mod.rs @@ -143,7 +143,6 @@ impl PublicationResult { /// A PublishHandler is a Handler which publishes catalog specifications. #[derive(Clone)] -#[allow(dead_code)] pub struct Publisher { flowctl_go: std::path::PathBuf, builds_root: url::Url, diff --git a/crates/control-plane-api/src/server/create_data_plane.rs b/crates/control-plane-api/src/server/create_data_plane.rs index 243856f2b8b..0615d63b5a4 100644 --- a/crates/control-plane-api/src/server/create_data_plane.rs +++ b/crates/control-plane-api/src/server/create_data_plane.rs @@ -63,14 +63,15 @@ pub async fn create_data_plane( category, }): super::Request, ) -> Result, crate::server::error::ApiError> { - let models::authorizations::ControlClaims { sub: user_id, .. } = env.claims()?; + let claims = env.claims()?; + let user_id = &claims.sub; + let user_email = claims.email.as_deref().unwrap_or("user"); - if !user_can_admin_ops(&app.snapshot_watch, *user_id) { - return Err(tonic::Status::permission_denied( - "authenticated user is not an admin of the 'ops/' tenant", - ) - .into()); - } + // Authorize against the request's Snapshot (not a fresh watch token) so that + // `authorization_outcome` can refresh-and-retry a denial that was decided + // from a Snapshot older than the request. + let policy_result = evaluate_ops_admin_authorization(env.snapshot(), *user_id, user_email); + let (_expiry, ()) = env.authorization_outcome(policy_result).await?; let (data_plane_fqdn, base_name, pulumi_stack) = match &private { None => ( @@ -311,26 +312,37 @@ pub async fn create_data_plane( Ok(axum::Json(Response {})) } -/// Returns whether `user_id` holds `admin` capability resolving to exactly the -/// `ops/` tenant, which is required to create a data-plane. +/// Builds a policy result (for `Envelope::authorization_outcome`) asserting that +/// `user_id` holds `admin` capability resolving to exactly the `ops/` tenant, +/// which is required to create a data-plane. /// -/// This is evaluated against the authorization `Snapshot` (as was done for the -/// storage-mappings directive), replacing the prior +/// This replaces the prior /// `internal.user_roles($user, 'admin') where role_prefix = 'ops/'` SQL check. /// Because that check was an *exact* `ops/` match — a data-plane always lives /// under `ops/` — this looks for `admin` at precisely the `ops/` prefix, and /// deliberately does not accept a grant at an ancestor or a descendant of it. -pub(super) fn user_can_admin_ops( - snapshot_watch: &std::sync::Arc>, +/// +/// It evaluates against a caller-provided `Snapshot` (the request's Snapshot via +/// `Envelope::snapshot`) rather than re-reading the watch, so that +/// `authorization_outcome` can refresh-and-retry a denial decided from a stale +/// Snapshot. +pub(super) fn evaluate_ops_admin_authorization( + snapshot: &crate::Snapshot, user_id: uuid::Uuid, -) -> bool { - let refresh = snapshot_watch.token(); - let snapshot = refresh.result().unwrap(); - - snapshot + user_email: &str, +) -> crate::AuthZResult<()> { + let is_ops_admin = snapshot .prefix_and_capabilities_per_user(user_id) .get("ops/") - .is_some_and(|(_bits, legacy)| *legacy >= models::Capability::Admin) + .is_some_and(|(_bits, legacy)| *legacy >= models::Capability::Admin); + + if is_ops_admin { + Ok((None, ())) + } else { + Err(tonic::Status::permission_denied(format!( + "{user_email} is not an admin of the 'ops/' tenant", + ))) + } } impl Validate for Category { @@ -345,7 +357,7 @@ impl Validate for Category { #[cfg(test)] mod test { - use super::user_can_admin_ops; + use super::evaluate_ops_admin_authorization; // Inserts `user_id` along with the given `user_grants` // (as `(object_role, capability)`) and `role_grants` (as @@ -396,7 +408,9 @@ mod test { // snapshot built from the grants inserted above, rather than the empty // snapshot used to exercise the server's retry flow. let snapshot_watch = crate::test_server::snapshot(pool.clone(), false).await; - user_can_admin_ops(&snapshot_watch, user_id) + let refresh = snapshot_watch.token(); + let snapshot = refresh.result().unwrap(); + evaluate_ops_admin_authorization(snapshot, user_id, "test@example.com").is_ok() } // A user granted `admin` directly on `ops/` is authorized. diff --git a/crates/control-plane-api/src/server/update_l2_reporting.rs b/crates/control-plane-api/src/server/update_l2_reporting.rs index 445560d2234..b69a70a01c8 100644 --- a/crates/control-plane-api/src/server/update_l2_reporting.rs +++ b/crates/control-plane-api/src/server/update_l2_reporting.rs @@ -32,14 +32,19 @@ pub async fn update_l2_reporting( dry_run, }): super::Request, ) -> Result, crate::ApiError> { - let crate::ControlClaims { sub: user_id, .. } = env.claims()?; - - if !super::create_data_plane::user_can_admin_ops(&app.snapshot_watch, *user_id) { - return Err(tonic::Status::permission_denied( - "authenticated user is not an admin of the 'ops/' tenant", - ) - .into()); - } + let claims = env.claims()?; + let user_id = &claims.sub; + let user_email = claims.email.as_deref().unwrap_or("user"); + + // Authorize against the request's Snapshot (not a fresh watch token) so that + // `authorization_outcome` can refresh-and-retry a denial that was decided + // from a Snapshot older than the request. + let policy_result = super::create_data_plane::evaluate_ops_admin_authorization( + env.snapshot(), + *user_id, + user_email, + ); + let (_expiry, ()) = env.authorization_outcome(policy_result).await?; let template = include_str!("../../../../ops-catalog/reporting-L2-template.bundle.json"); let tables::DraftCatalog { collections, .. } = From e315c31236ad1499126a213f72a9331ddb8fba5b Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Tue, 21 Jul 2026 16:23:53 +0000 Subject: [PATCH 08/12] Did a refactoring to improve evaluation speed of test harness tests. --- crates/agent/src/integration_tests/harness.rs | 53 +++++++++++++------ 1 file changed, 37 insertions(+), 16 deletions(-) diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index c9641806cec..6fa973cdef5 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -180,6 +180,12 @@ pub struct TestHarness { /// Live authorization Snapshot watch, retained so tests can force it to /// re-fetch from Postgres after mutating grants. See `refresh_snapshot`. pub snapshot_watch: Arc>, + /// Write handle for `snapshot_watch`: pushes a freshly-fetched Snapshot into + /// the same watch. The harness drives Snapshot refreshes explicitly (see + /// `refresh_snapshot`) rather than through `PgSnapshotSource`'s timer-gated + /// polling loop, which would otherwise impose a `MIN_REFRESH_INTERVAL` + /// cool-off on every refresh. + set_snapshot: Box, #[allow(dead_code)] // only here so we don't drop it until the harness is dropped pub builds_root: tempfile::TempDir, pub discover_handler: DiscoverHandler, @@ -242,8 +248,19 @@ impl HarnessBuilder { eprintln!("end of PUB-LOG"); }); - let snapshot_source = control_plane_api::snapshot::PgSnapshotSource::new(pool.clone()); - let snapshot_watch = tokens::watch(snapshot_source).ready_owned().await; + // Back the authorization Snapshot with a manually-driven watch rather + // than `PgSnapshotSource`'s polling loop. Tests never refresh on a timer; + // they push a freshly-fetched Snapshot via `set_snapshot` whenever they + // mutate grants (see `refresh_snapshot`), which avoids the source's + // `MIN_REFRESH_INTERVAL` cool-off blocking the (real-time) test clock. + let (snapshot_pending, snapshot_replace) = + tokens::manual::(); + let set_snapshot: Box = + Box::new(move |snapshot| { + _ = snapshot_replace(Ok(snapshot)); + }); + set_snapshot(TestHarness::fetch_snapshot(&pool).await); + let snapshot_watch = snapshot_pending.ready_owned().await; let mock_connectors = connectors::MockDiscoverConnectors::default(); let discover_handler = @@ -287,6 +304,7 @@ impl HarnessBuilder { pool, publisher, snapshot_watch, + set_snapshot, builds_root, discover_handler, control_plane, @@ -555,23 +573,26 @@ impl TestHarness { &mut self.control_plane } + /// Fetches the current authorization state from Postgres and builds a + /// Snapshot from it. This performs the same query `PgSnapshotSource` runs, + /// but without its `MIN_REFRESH_INTERVAL` cool-off, so the harness can + /// refresh synchronously and deterministically. + async fn fetch_snapshot(pool: &sqlx::PgPool) -> control_plane_api::Snapshot { + let mut decrypted_hmac_keys = std::collections::HashMap::new(); + let data = control_plane_api::snapshot::try_fetch(pool, &mut decrypted_hmac_keys) + .await + .expect("failed to fetch authorization snapshot"); + control_plane_api::Snapshot::new(tokens::now(), data) + } + /// Forces the in-memory authorization Snapshot to re-fetch from Postgres, so /// that grant changes written directly to the DB become visible to publication - /// authorization. Integration tests run with paused time and never refresh the - /// Snapshot automatically, so grant-mutating helpers call this explicitly. + /// authorization. Tests never refresh the Snapshot on a timer, so + /// grant-mutating helpers call this explicitly to push the fresh state into + /// `snapshot_watch`. pub async fn refresh_snapshot(&self) { - let current = self.snapshot_watch.token(); - let Ok(snapshot) = current.result() else { - return; // No live Snapshot to revoke; nothing to refresh. - }; - let prev_version = current.version(); - // Cancelling `revoke` signals `PgSnapshotSource` to re-fetch immediately, - // even under paused test time (the trigger is cancellation, not a timer). - snapshot.revoke.cancel(); - // Wait until the watch publishes the newer, re-fetched Snapshot. - while self.snapshot_watch.version() == prev_version { - tokio::task::yield_now().await; - } + let snapshot = Self::fetch_snapshot(&self.pool).await; + (self.set_snapshot)(snapshot); } /// Setup a new tenant with the given name, and return the id of the user From fba4815d8dfbdf0f584fbb5f7194c79276504a8c Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Wed, 22 Jul 2026 15:46:55 +0000 Subject: [PATCH 09/12] Did a refactoring post merge. --- ...9de98bc61be2843033a74a9bf0c5c9d44c410.json | 2 +- ...ca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json | 188 ------------------ ...78286d1024dd706152c005983637982365277.json | 2 +- .../.tenant_alerts.rs.pending-snap | 10 + .../.user_publications.rs.pending-snap | 8 + .../control-plane-api/src/evolutions/mod.rs | 3 +- crates/control-plane-api/src/live_specs/db.rs | 49 +++-- .../control-plane-api/src/live_specs/mod.rs | 7 +- .../src/publications/initialize.rs | 5 +- 9 files changed, 55 insertions(+), 219 deletions(-) delete mode 100644 .sqlx/query-d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json diff --git a/.sqlx/query-b10fda873a90630129d968ed7369de98bc61be2843033a74a9bf0c5c9d44c410.json b/.sqlx/query-b10fda873a90630129d968ed7369de98bc61be2843033a74a9bf0c5c9d44c410.json index 32826828169..4828299c4ee 100644 --- a/.sqlx/query-b10fda873a90630129d968ed7369de98bc61be2843033a74a9bf0c5c9d44c410.json +++ b/.sqlx/query-b10fda873a90630129d968ed7369de98bc61be2843033a74a9bf0c5c9d44c410.json @@ -95,7 +95,7 @@ false, true, false, - true, + false, false, false, true, diff --git a/.sqlx/query-d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json b/.sqlx/query-d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json deleted file mode 100644 index 0fe92b4b2c0..00000000000 --- a/.sqlx/query-d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4.json +++ /dev/null @@ -1,188 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n with collections(id) as (\n select ls.id\n from unnest($1::text[]) as names(catalog_name)\n join live_specs ls on ls.catalog_name = names.catalog_name\n ),\n exp(id) as (\n select lsf.source_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.target_id\n union\n select lsf.target_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.source_id\n ),\n user_roles as materialized (\n select role_prefix, capability from UNNEST($3::text[], $4::grant_capability[]) as t(role_prefix, capability)\n )\n select\n ls.id as \"id: Id\",\n ls.last_pub_id as \"last_pub_id: Id\",\n ls.last_build_id as \"last_build_id: Id\",\n ls.data_plane_id as \"data_plane_id: Id\",\n ls.catalog_name,\n ls.spec_type as \"spec_type?: CatalogType\",\n ls.spec as \"spec: TextJson>\",\n ls.built_spec as \"built_spec: TextJson>\",\n ls.inferred_schema_md5,\n (\n select max(capability) from user_roles r\n where starts_with(ls.catalog_name, r.role_prefix)\n ) as \"user_capability: Capability\",\n coalesce(\n (select json_agg(row_to_json(role_grants))\n from role_grants\n where starts_with(ls.catalog_name, subject_role)),\n '[]'\n ) as \"spec_capabilities!: Json>\",\n ls.dependency_hash\n from exp\n join live_specs ls on ls.id = exp.id\n where ls.spec is not null and not ls.catalog_name = any($2);\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 1, - "name": "last_pub_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 2, - "name": "last_build_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 3, - "name": "data_plane_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 4, - "name": "catalog_name", - "type_info": "Text" - }, - { - "ordinal": 5, - "name": "spec_type?: CatalogType", - "type_info": { - "Custom": { - "name": "catalog_spec_type", - "kind": { - "Enum": [ - "capture", - "collection", - "materialization", - "test" - ] - } - } - } - }, - { - "ordinal": 6, - "name": "spec: TextJson>", - "type_info": "Json" - }, - { - "ordinal": 7, - "name": "built_spec: TextJson>", - "type_info": "Json" - }, - { - "ordinal": 8, - "name": "inferred_schema_md5", - "type_info": "Text" - }, - { - "ordinal": 9, - "name": "user_capability: Capability", - "type_info": { - "Custom": { - "name": "grant_capability", - "kind": { - "Enum": [ - "none", - "x_01", - "x_02", - "x_03", - "x_04", - "x_05", - "x_06", - "x_07", - "x_08", - "x_09", - "read", - "x_11", - "x_12", - "x_13", - "x_14", - "x_15", - "x_16", - "x_17", - "x_18", - "x_19", - "write", - "x_21", - "x_22", - "x_23", - "x_24", - "x_25", - "x_26", - "x_27", - "x_28", - "x_29", - "admin" - ] - } - } - } - }, - { - "ordinal": 10, - "name": "spec_capabilities!: Json>", - "type_info": "Json" - }, - { - "ordinal": 11, - "name": "dependency_hash", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "TextArray", - "TextArray", - "TextArray", - { - "Custom": { - "name": "grant_capability[]", - "kind": { - "Array": { - "Custom": { - "name": "grant_capability", - "kind": { - "Enum": [ - "none", - "x_01", - "x_02", - "x_03", - "x_04", - "x_05", - "x_06", - "x_07", - "x_08", - "x_09", - "read", - "x_11", - "x_12", - "x_13", - "x_14", - "x_15", - "x_16", - "x_17", - "x_18", - "x_19", - "write", - "x_21", - "x_22", - "x_23", - "x_24", - "x_25", - "x_26", - "x_27", - "x_28", - "x_29", - "admin" - ] - } - } - } - } - } - } - ] - }, - "nullable": [ - false, - false, - false, - false, - false, - true, - true, - true, - true, - null, - null, - true - ] - }, - "hash": "d9df18630499408fa85430eeaadca2e4a9ec6cce6a62227a29c2f841ceea1bd4" -} diff --git a/.sqlx/query-e9c8d33f3b85a5964538fc00fa678286d1024dd706152c005983637982365277.json b/.sqlx/query-e9c8d33f3b85a5964538fc00fa678286d1024dd706152c005983637982365277.json index 8f9e316830b..f76bd045e58 100644 --- a/.sqlx/query-e9c8d33f3b85a5964538fc00fa678286d1024dd706152c005983637982365277.json +++ b/.sqlx/query-e9c8d33f3b85a5964538fc00fa678286d1024dd706152c005983637982365277.json @@ -64,7 +64,7 @@ false, null, false, - true, + false, true, false, null, diff --git a/crates/agent/src/integration_tests/.tenant_alerts.rs.pending-snap b/crates/agent/src/integration_tests/.tenant_alerts.rs.pending-snap index 26d6e544f1f..566be0fd902 100644 --- a/crates/agent/src/integration_tests/.tenant_alerts.rs.pending-snap +++ b/crates/agent/src/integration_tests/.tenant_alerts.rs.pending-snap @@ -14,3 +14,13 @@ {"run_id":"1784726685-655096928","line":91,"new":null,"old":null} {"run_id":"1784726685-655096928","line":137,"new":null,"old":null} {"run_id":"1784726685-655096928","line":181,"new":null,"old":null} +{"run_id":"1784732770-555533510","line":19,"new":null,"old":null} +{"run_id":"1784732770-555533510","line":51,"new":null,"old":null} +{"run_id":"1784732770-555533510","line":91,"new":null,"old":null} +{"run_id":"1784732770-555533510","line":137,"new":null,"old":null} +{"run_id":"1784732770-555533510","line":181,"new":null,"old":null} +{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":19,"new":null,"old":null} +{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":51,"new":null,"old":null} +{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":91,"new":null,"old":null} +{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":137,"new":null,"old":null} +{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":181,"new":null,"old":null} diff --git a/crates/agent/src/integration_tests/.user_publications.rs.pending-snap b/crates/agent/src/integration_tests/.user_publications.rs.pending-snap index adc0e6ab600..18e519929ed 100644 --- a/crates/agent/src/integration_tests/.user_publications.rs.pending-snap +++ b/crates/agent/src/integration_tests/.user_publications.rs.pending-snap @@ -5,3 +5,11 @@ {"run_id":"1784726664-40620475","line":167,"new":null,"old":null} {"run_id":"1784726685-655096928","line":142,"new":null,"old":null} {"run_id":"1784726685-655096928","line":167,"new":null,"old":null} +{"run_id":"1784732711-231719153","line":142,"new":null,"old":null} +{"run_id":"1784732711-231719153","line":167,"new":null,"old":null} +{"run_id":"1784732736-313546420","line":142,"new":null,"old":null} +{"run_id":"1784732736-313546420","line":167,"new":null,"old":null} +{"run_id":"1784732770-555533510","line":142,"new":null,"old":null} +{"run_id":"1784732770-555533510","line":167,"new":null,"old":null} +{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":142,"new":null,"old":null} +{"run_id":"43271747-75c0-4204-8de0-038686b68937","line":167,"new":null,"old":null} diff --git a/crates/control-plane-api/src/evolutions/mod.rs b/crates/control-plane-api/src/evolutions/mod.rs index ebf2c02ace3..2b4e1bc9a17 100644 --- a/crates/control-plane-api/src/evolutions/mod.rs +++ b/crates/control-plane-api/src/evolutions/mod.rs @@ -185,11 +185,12 @@ pub async fn evolve( .collect::>(); let exclude_names = draft.all_spec_names().collect::>(); let expanded_live = crate::live_specs::get_connected_live_specs( + user_id, &collection_names, &exclude_names, capability_filter, db, - &prefixes_and_capabilities, + &snapshot, ) .await?; draft.add_live(expanded_live); diff --git a/crates/control-plane-api/src/live_specs/db.rs b/crates/control-plane-api/src/live_specs/db.rs index 352ea07dbc9..1d46f910afb 100644 --- a/crates/control-plane-api/src/live_specs/db.rs +++ b/crates/control-plane-api/src/live_specs/db.rs @@ -134,16 +134,13 @@ pub async fn fetch_inferred_schemas( /// Queries for all non-deleted `live_specs` that are connected to the given `collection_names` via /// `live_spec_flows`. pub async fn fetch_expanded_live_specs( + user_id: uuid::Uuid, collection_names: &[&str], exclude_names: &[&str], db: impl sqlx::Executor<'_, Database = sqlx::Postgres>, - permissions_set: &PrefixesAndCapabilities<'_>, + snapshot: &crate::Snapshot, ) -> sqlx::Result> { - let (prefixes, capabilities): (Vec, Vec) = permissions_set - .iter() - .map(|(prefix, capabilities)| (prefix.to_string(), capabilities.1)) - .unzip(); - sqlx::query_as!( + let mut expanded = sqlx::query_as!( LiveSpec, r#" with collections(id) as ( @@ -159,9 +156,6 @@ pub async fn fetch_expanded_live_specs( select lsf.target_id as id from collections c join live_spec_flows lsf on c.id = lsf.source_id - ), - user_roles as materialized ( - select role_prefix, capability from UNNEST($3::text[], $4::grant_capability[]) as t(role_prefix, capability) ) select ls.id as "id: Id", @@ -173,16 +167,11 @@ pub async fn fetch_expanded_live_specs( ls.spec as "spec: TextJson>", ls.built_spec as "built_spec: TextJson>", ls.inferred_schema_md5, - ( - select max(capability) from user_roles r - where starts_with(ls.catalog_name, r.role_prefix) - ) as "user_capability: Capability", - coalesce( - (select json_agg(row_to_json(role_grants)) - from role_grants - where starts_with(ls.catalog_name, subject_role)), - '[]' - ) as "spec_capabilities!: Json>", + -- `user_capability` and `spec_capabilities` are synthesized from the + -- authorization Snapshot below rather than queried here. The `null` + -- (→ None) and empty-array placeholders are overwritten afterwards. + null as "user_capability: Capability", + '[]' as "spec_capabilities!: Json>", ls.dependency_hash from exp join live_specs ls on ls.id = exp.id @@ -190,11 +179,27 @@ pub async fn fetch_expanded_live_specs( "#, collection_names as &[&str], exclude_names as &[&str], - &prefixes, - &capabilities as &[Capability], ) .fetch_all(db) - .await + .await?; + + // Compute each spec's `user_capability` (the user's greatest capability among + // the prefixes that `catalog_name` falls under) and `spec_capabilities` (the + // role grants the spec holds by virtue of its own name/role) from the + // Snapshot, replacing the previous `user_roles` and `role_grants` subqueries. + let reachable = snapshot.prefix_and_capabilities_per_user(user_id); + for spec in expanded.iter_mut() { + let mut max_capability: Option = None; + for (prefix, (_, capability)) in reachable.iter() { + if spec.catalog_name.starts_with(*prefix) { + max_capability = max_capability.max(Some(*capability)); + } + } + spec.user_capability = max_capability; + spec.spec_capabilities = Json(snapshot.spec_capabilities(&spec.catalog_name)); + } + + Ok(expanded) } /// Returns all live spec names under the given prefix. diff --git a/crates/control-plane-api/src/live_specs/mod.rs b/crates/control-plane-api/src/live_specs/mod.rs index 0ae8a09cfeb..ce65a86a4d7 100644 --- a/crates/control-plane-api/src/live_specs/mod.rs +++ b/crates/control-plane-api/src/live_specs/mod.rs @@ -7,7 +7,6 @@ pub use db::{ }; use models::Capability; use std::ops::Deref; -use uuid::Uuid; /// Fetches live specs, returning them as a `tables::LiveCatalog`. Optionally /// filters the specs based on user capability. If `filter_capability` is @@ -76,14 +75,16 @@ pub async fn get_live_specs( } pub async fn get_connected_live_specs( + user_id: uuid::Uuid, collection_names: &[&str], exclude_names: &[&str], filter_capability: Option, db: &sqlx::PgPool, - permissions_set: &PrefixesAndCapabilities<'_>, + snapshot: &crate::Snapshot, ) -> anyhow::Result { let expanded_rows = - db::fetch_expanded_live_specs(collection_names, exclude_names, db, permissions_set).await?; + db::fetch_expanded_live_specs(user_id, collection_names, exclude_names, db, snapshot) + .await?; let mut live = tables::LiveCatalog::default(); for exp in expanded_rows { if let Some(minimum_capability) = filter_capability { diff --git a/crates/control-plane-api/src/publications/initialize.rs b/crates/control-plane-api/src/publications/initialize.rs index ef554522c7d..e971358e675 100644 --- a/crates/control-plane-api/src/publications/initialize.rs +++ b/crates/control-plane-api/src/publications/initialize.rs @@ -92,15 +92,14 @@ impl Initialize for ExpandDraft { }; let snapshot = snapshot_watch.token(); let snapshot = snapshot.result().unwrap(); - let prefixes_and_capabilities = snapshot.prefix_and_capabilities_per_user(user_id); let expanded_catalog = crate::live_specs::get_connected_live_specs( - // user_id, + user_id, &drafted_collections, &all_drafted_specs, capability_filter, db, - &prefixes_and_capabilities, + snapshot, ) .await?; tracing::debug!( From e78e852125784c3ee7520ba19277d2eba63fb16f Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Wed, 22 Jul 2026 15:47:04 +0000 Subject: [PATCH 10/12] Missed a file. --- ...6666634434ea11938b6d0c655b7e2e8054bf1.json | 101 ++++++++++++++++++ 1 file changed, 101 insertions(+) create mode 100644 .sqlx/query-74d603c7cacef879838e1573a866666634434ea11938b6d0c655b7e2e8054bf1.json diff --git a/.sqlx/query-74d603c7cacef879838e1573a866666634434ea11938b6d0c655b7e2e8054bf1.json b/.sqlx/query-74d603c7cacef879838e1573a866666634434ea11938b6d0c655b7e2e8054bf1.json new file mode 100644 index 00000000000..88031d307dd --- /dev/null +++ b/.sqlx/query-74d603c7cacef879838e1573a866666634434ea11938b6d0c655b7e2e8054bf1.json @@ -0,0 +1,101 @@ +{ + "db_name": "PostgreSQL", + "query": "\n with collections(id) as (\n select ls.id\n from unnest($1::text[]) as names(catalog_name)\n join live_specs ls on ls.catalog_name = names.catalog_name\n ),\n exp(id) as (\n select lsf.source_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.target_id\n union\n select lsf.target_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.source_id\n )\n select\n ls.id as \"id: Id\",\n ls.last_pub_id as \"last_pub_id: Id\",\n ls.last_build_id as \"last_build_id: Id\",\n ls.data_plane_id as \"data_plane_id: Id\",\n ls.catalog_name,\n ls.spec_type as \"spec_type?: CatalogType\",\n ls.spec as \"spec: TextJson>\",\n ls.built_spec as \"built_spec: TextJson>\",\n ls.inferred_schema_md5,\n -- `user_capability` and `spec_capabilities` are synthesized from the\n -- authorization Snapshot below rather than queried here. The `null`\n -- (→ None) and empty-array placeholders are overwritten afterwards.\n null as \"user_capability: Capability\",\n '[]' as \"spec_capabilities!: Json>\",\n ls.dependency_hash\n from exp\n join live_specs ls on ls.id = exp.id\n where ls.spec is not null and not ls.catalog_name = any($2);\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "last_pub_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 2, + "name": "last_build_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 3, + "name": "data_plane_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 4, + "name": "catalog_name", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "spec_type?: CatalogType", + "type_info": { + "Custom": { + "name": "catalog_spec_type", + "kind": { + "Enum": [ + "capture", + "collection", + "materialization", + "test" + ] + } + } + } + }, + { + "ordinal": 6, + "name": "spec: TextJson>", + "type_info": "Json" + }, + { + "ordinal": 7, + "name": "built_spec: TextJson>", + "type_info": "Json" + }, + { + "ordinal": 8, + "name": "inferred_schema_md5", + "type_info": "Text" + }, + { + "ordinal": 9, + "name": "user_capability: Capability", + "type_info": "Text" + }, + { + "ordinal": 10, + "name": "spec_capabilities!: Json>", + "type_info": "Text" + }, + { + "ordinal": 11, + "name": "dependency_hash", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "TextArray", + "TextArray" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + true, + true, + true, + true, + null, + null, + true + ] + }, + "hash": "74d603c7cacef879838e1573a866666634434ea11938b6d0c655b7e2e8054bf1" +} From 22e181bc50ed5eca04f4008660432082e43681c0 Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Wed, 22 Jul 2026 16:15:20 +0000 Subject: [PATCH 11/12] Did refactoring based on a review on another related PR. --- ...41840a1743da8a5d1863156d6207853b19186.json | 119 ------------------ crates/control-plane-api/src/evolutions/db.rs | 58 +++++---- 2 files changed, 36 insertions(+), 141 deletions(-) delete mode 100644 .sqlx/query-11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186.json diff --git a/.sqlx/query-11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186.json b/.sqlx/query-11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186.json deleted file mode 100644 index 939eeec2dd0..00000000000 --- a/.sqlx/query-11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186.json +++ /dev/null @@ -1,119 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n with user_roles as materialized (\n select role_prefix, capability from UNNEST($3::text[], $4::grant_capability[]) as t(role_prefix, capability)\n ),\n drafted as (\n select\n ds.catalog_name,\n ds.id as draft_spec_id,\n ls.id as live_spec_id,\n ds.expect_pub_id,\n ls.last_pub_id as last_pub_id,\n ds.spec as spec,\n ds.spec_type as spec_type\n from draft_specs ds\n left join live_specs ls\n on ds.catalog_name = ls.catalog_name\n -- filter out live_specs rows that the user does not have admin access to\n and exists (select 1 from user_roles r where ls.catalog_name ^@ r.role_prefix)\n where ds.draft_id = $1\n ),\n not_drafted as (\n select catalog_name from unnest($2::text[]) as names(catalog_name)\n except\n select catalog_name from drafted\n ),\n live as (\n select\n ls.catalog_name,\n ls.spec,\n ls.spec_type,\n ls.last_pub_id,\n ls.id\n from not_drafted\n join live_specs ls on not_drafted.catalog_name = ls.catalog_name\n where\n -- filter out live_specs rows that the user does not have admin access to\n exists (select 1 from user_roles r where ls.catalog_name ^@ r.role_prefix)\n )\n select\n catalog_name as \"catalog_name!: String\",\n draft_spec_id as \"draft_spec_id: Id\",\n live_spec_id as \"live_spec_id: Id\",\n expect_pub_id as \"expect_pub_id: Id\",\n last_pub_id as \"last_pub_id: Id\",\n spec as \"spec: Json>\",\n spec_type as \"spec_type: CatalogType\"\n from drafted\n union all\n select\n catalog_name as \"catalog_name!: String\",\n null as \"draft_spec_id: Id\",\n id as \"live_spec_id: Id\",\n null as \"expect_pub_id: Id\",\n last_pub_id as \"last_pub_id: Id\",\n spec as \"spec: Json>\",\n spec_type as \"spec_type: CatalogType\"\n from live\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "catalog_name!: String", - "type_info": "Text" - }, - { - "ordinal": 1, - "name": "draft_spec_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 2, - "name": "live_spec_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 3, - "name": "expect_pub_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 4, - "name": "last_pub_id: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 5, - "name": "spec: Json>", - "type_info": "Json" - }, - { - "ordinal": 6, - "name": "spec_type: CatalogType", - "type_info": { - "Custom": { - "name": "catalog_spec_type", - "kind": { - "Enum": [ - "capture", - "collection", - "materialization", - "test" - ] - } - } - } - } - ], - "parameters": { - "Left": [ - "Macaddr8", - "TextArray", - "TextArray", - { - "Custom": { - "name": "grant_capability[]", - "kind": { - "Array": { - "Custom": { - "name": "grant_capability", - "kind": { - "Enum": [ - "none", - "x_01", - "x_02", - "x_03", - "x_04", - "x_05", - "x_06", - "x_07", - "x_08", - "x_09", - "read", - "x_11", - "x_12", - "x_13", - "x_14", - "x_15", - "x_16", - "x_17", - "x_18", - "x_19", - "write", - "x_21", - "x_22", - "x_23", - "x_24", - "x_25", - "x_26", - "x_27", - "x_28", - "x_29", - "admin" - ] - } - } - } - } - } - } - ] - }, - "nullable": [ - null, - null, - null, - null, - null, - null, - null - ] - }, - "hash": "11ea317b38e527c944bf58ef38e41840a1743da8a5d1863156d6207853b19186" -} diff --git a/crates/control-plane-api/src/evolutions/db.rs b/crates/control-plane-api/src/evolutions/db.rs index 833d4b7139e..dc8b447161b 100644 --- a/crates/control-plane-api/src/evolutions/db.rs +++ b/crates/control-plane-api/src/evolutions/db.rs @@ -1,4 +1,4 @@ -use crate::{TextJson as Json, snapshot::PrefixesAndCapabilities}; +use crate::TextJson as Json; use chrono::{DateTime, Utc}; use models::{Capability, CatalogType, Id}; use serde::Serialize; @@ -81,22 +81,14 @@ pub struct SpecRow { pub async fn resolve_specs( draft_id: Id, collection_names: Vec, - permissions_set: &PrefixesAndCapabilities<'_>, + user_id: uuid::Uuid, + snapshot: &crate::Snapshot, txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, ) -> sqlx::Result> { - // Doing the filter before handing this off to the query to reduce the amount of data being sent to the db. - let (prefixes, capabilities): (Vec, Vec) = permissions_set - .iter() - .filter(|(_prefix, capabilities)| capabilities.1 >= Capability::Admin) - .map(|(prefix, capabilities)| (prefix.to_string(), capabilities.1)) - .unzip(); - sqlx::query_as!( + let mut rows = sqlx::query_as!( SpecRow, r#" - with user_roles as materialized ( - select role_prefix, capability from UNNEST($3::text[], $4::grant_capability[]) as t(role_prefix, capability) - ), - drafted as ( + with drafted as ( select ds.catalog_name, ds.id as draft_spec_id, @@ -108,8 +100,6 @@ pub async fn resolve_specs( from draft_specs ds left join live_specs ls on ds.catalog_name = ls.catalog_name - -- filter out live_specs rows that the user does not have admin access to - and exists (select 1 from user_roles r where ls.catalog_name ^@ r.role_prefix) where ds.draft_id = $1 ), not_drafted as ( @@ -126,9 +116,6 @@ pub async fn resolve_specs( ls.id from not_drafted join live_specs ls on not_drafted.catalog_name = ls.catalog_name - where - -- filter out live_specs rows that the user does not have admin access to - exists (select 1 from user_roles r where ls.catalog_name ^@ r.role_prefix) ) select catalog_name as "catalog_name!: String", @@ -152,10 +139,37 @@ pub async fn resolve_specs( "#, draft_id as Id, collection_names as Vec, - &prefixes, - &capabilities as &[Capability], - ).fetch_all(&mut **txn) - .await + ) + .fetch_all(&mut **txn) + .await?; + + // The admin filter previously applied in SQL (an `^@` prefix match against + // the user's `>= Admin` role prefixes) is now applied here from the + // Snapshot. It has two distinct effects, both preserved: + // * a drafted spec is always retained, but its live-spec linkage + // (`live_spec_id`/`last_pub_id`) is cleared when the user lacks admin — + // mirroring the previous conditional LEFT JOIN; while + // * a live-only (not-drafted) spec is dropped entirely without admin. + let reachable = snapshot.prefix_and_capabilities_per_user(user_id); + let user_can_admin = |catalog_name: &str| { + reachable.iter().any(|(prefix, (_, legacy))| { + *legacy >= Capability::Admin && catalog_name.starts_with(*prefix) + }) + }; + + rows.retain_mut(|row| { + if row.draft_spec_id.is_some() { + if !user_can_admin(&row.catalog_name) { + row.live_spec_id = None; + row.last_pub_id = None; + } + true + } else { + user_can_admin(&row.catalog_name) + } + }); + + Ok(rows) } pub async fn fetch_resource_spec_schema( From 68eecebaba6cadf613d9ac61f96157049b0734c0 Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Wed, 22 Jul 2026 16:15:34 +0000 Subject: [PATCH 12/12] Missed a file. --- ...e0acaca3d87b997e02da94513b584a2a0d6ce.json | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 .sqlx/query-6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce.json diff --git a/.sqlx/query-6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce.json b/.sqlx/query-6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce.json new file mode 100644 index 00000000000..9246b22b475 --- /dev/null +++ b/.sqlx/query-6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce.json @@ -0,0 +1,71 @@ +{ + "db_name": "PostgreSQL", + "query": "\n with drafted as (\n select\n ds.catalog_name,\n ds.id as draft_spec_id,\n ls.id as live_spec_id,\n ds.expect_pub_id,\n ls.last_pub_id as last_pub_id,\n ds.spec as spec,\n ds.spec_type as spec_type\n from draft_specs ds\n left join live_specs ls\n on ds.catalog_name = ls.catalog_name\n where ds.draft_id = $1\n ),\n not_drafted as (\n select catalog_name from unnest($2::text[]) as names(catalog_name)\n except\n select catalog_name from drafted\n ),\n live as (\n select\n ls.catalog_name,\n ls.spec,\n ls.spec_type,\n ls.last_pub_id,\n ls.id\n from not_drafted\n join live_specs ls on not_drafted.catalog_name = ls.catalog_name\n )\n select\n catalog_name as \"catalog_name!: String\",\n draft_spec_id as \"draft_spec_id: Id\",\n live_spec_id as \"live_spec_id: Id\",\n expect_pub_id as \"expect_pub_id: Id\",\n last_pub_id as \"last_pub_id: Id\",\n spec as \"spec: Json>\",\n spec_type as \"spec_type: CatalogType\"\n from drafted\n union all\n select\n catalog_name as \"catalog_name!: String\",\n null as \"draft_spec_id: Id\",\n id as \"live_spec_id: Id\",\n null as \"expect_pub_id: Id\",\n last_pub_id as \"last_pub_id: Id\",\n spec as \"spec: Json>\",\n spec_type as \"spec_type: CatalogType\"\n from live\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "catalog_name!: String", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "draft_spec_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 2, + "name": "live_spec_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 3, + "name": "expect_pub_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 4, + "name": "last_pub_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 5, + "name": "spec: Json>", + "type_info": "Json" + }, + { + "ordinal": 6, + "name": "spec_type: CatalogType", + "type_info": { + "Custom": { + "name": "catalog_spec_type", + "kind": { + "Enum": [ + "capture", + "collection", + "materialization", + "test" + ] + } + } + } + } + ], + "parameters": { + "Left": [ + "Macaddr8", + "TextArray" + ] + }, + "nullable": [ + null, + null, + null, + null, + null, + null, + null + ] + }, + "hash": "6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce" +}