diff --git a/.sqlx/query-51e1e6a17507c4c857e26536cb6dc7e4cac95ef04f4f4ec81d3459e01589395e.json b/.sqlx/query-951e3d24ce062de78dc96dd20da51ee0d5b77c19e41a2ae62d08c9084b20a492.json similarity index 75% rename from .sqlx/query-51e1e6a17507c4c857e26536cb6dc7e4cac95ef04f4f4ec81d3459e01589395e.json rename to .sqlx/query-951e3d24ce062de78dc96dd20da51ee0d5b77c19e41a2ae62d08c9084b20a492.json index a03cde1d529..ebdab6580f1 100644 --- a/.sqlx/query-51e1e6a17507c4c857e26536cb6dc7e4cac95ef04f4f4ec81d3459e01589395e.json +++ b/.sqlx/query-951e3d24ce062de78dc96dd20da51ee0d5b77c19e41a2ae62d08c9084b20a492.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select ls.catalog_name as \"name!: String\"\n from live_specs ls\n left outer join data_planes dp on ls.data_plane_id = dp.id\n where (coalesce(array_length($1::catalog_name[], 1), 0) = 0 or ls.catalog_name = any($1::catalog_name[]))\n and ($2::text is null or ls.catalog_name::text ^@ $2::text)\n and ($3::catalog_spec_type is null or ls.spec_type = $3::catalog_spec_type)\n and ($4::text is null or $4::text = dp.data_plane_name)\n and ($5::catalog_name is null or ls.catalog_name > $5::catalog_name)\n order by ls.catalog_name asc\n limit $6", + "query": "select ls.catalog_name as \"name!: String\"\n from live_specs ls\n left outer join data_planes dp on ls.data_plane_id = dp.id\n where (coalesce(array_length($1::catalog_name[], 1), 0) = 0 or ls.catalog_name = any($1::catalog_name[]))\n and ($2::text is null or ls.catalog_name::text ^@ $2::text)\n and ($3::catalog_spec_type is null or ls.spec_type = $3::catalog_spec_type)\n and ($4::text is null or $4::text = dp.data_plane_name)\n and (coalesce(array_length($5::text[], 1), 0) = 0 or ls.catalog_name::text = any($5::text[]))\n and (coalesce(array_length($6::text[], 1), 0) = 0 or ls.catalog_name::text ^@ any($6::text[]))\n and ($7::catalog_name is null or ls.catalog_name < $7::catalog_name)\n order by ls.catalog_name desc\n limit $8", "describe": { "columns": [ { @@ -41,6 +41,8 @@ } }, "Text", + "TextArray", + "TextArray", { "Custom": { "name": "catalog_name", @@ -56,5 +58,5 @@ false ] }, - "hash": "51e1e6a17507c4c857e26536cb6dc7e4cac95ef04f4f4ec81d3459e01589395e" + "hash": "951e3d24ce062de78dc96dd20da51ee0d5b77c19e41a2ae62d08c9084b20a492" } diff --git a/.sqlx/query-43283e37ab13116549678ab7dce0c65c0ab03291a07ac0665026be9f115f10cb.json b/.sqlx/query-e6c07a9353becd19ea45e821582ff17e0291a72060e4f36a166da5269b40b0f3.json similarity index 75% rename from .sqlx/query-43283e37ab13116549678ab7dce0c65c0ab03291a07ac0665026be9f115f10cb.json rename to .sqlx/query-e6c07a9353becd19ea45e821582ff17e0291a72060e4f36a166da5269b40b0f3.json index 0343017cc93..a3dddf5d91e 100644 --- a/.sqlx/query-43283e37ab13116549678ab7dce0c65c0ab03291a07ac0665026be9f115f10cb.json +++ b/.sqlx/query-e6c07a9353becd19ea45e821582ff17e0291a72060e4f36a166da5269b40b0f3.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "select ls.catalog_name as \"name!: String\"\n from live_specs ls\n left outer join data_planes dp on ls.data_plane_id = dp.id\n where (coalesce(array_length($1::catalog_name[], 1), 0) = 0 or ls.catalog_name = any($1::catalog_name[]))\n and ($2::text is null or ls.catalog_name::text ^@ $2::text)\n and ($3::catalog_spec_type is null or ls.spec_type = $3::catalog_spec_type)\n and ($4::text is null or $4::text = dp.data_plane_name)\n and ($5::catalog_name is null or ls.catalog_name < $5::catalog_name)\n order by ls.catalog_name desc\n limit $6", + "query": "select ls.catalog_name as \"name!: String\"\n from live_specs ls\n left outer join data_planes dp on ls.data_plane_id = dp.id\n where (coalesce(array_length($1::catalog_name[], 1), 0) = 0 or ls.catalog_name = any($1::catalog_name[]))\n and ($2::text is null or ls.catalog_name::text ^@ $2::text)\n and ($3::catalog_spec_type is null or ls.spec_type = $3::catalog_spec_type)\n and ($4::text is null or $4::text = dp.data_plane_name)\n and (coalesce(array_length($5::text[], 1), 0) = 0 or ls.catalog_name::text = any($5::text[]))\n and (coalesce(array_length($6::text[], 1), 0) = 0 or ls.catalog_name::text ^@ any($6::text[]))\n and ($7::catalog_name is null or ls.catalog_name > $7::catalog_name)\n order by ls.catalog_name asc\n limit $8", "describe": { "columns": [ { @@ -41,6 +41,8 @@ } }, "Text", + "TextArray", + "TextArray", { "Custom": { "name": "catalog_name", @@ -56,5 +58,5 @@ false ] }, - "hash": "43283e37ab13116549678ab7dce0c65c0ab03291a07ac0665026be9f115f10cb" + "hash": "e6c07a9353becd19ea45e821582ff17e0291a72060e4f36a166da5269b40b0f3" } diff --git a/crates/agent/src/integration_tests/graphql/queries/live_specs_filter_catalog_name.graphql b/crates/agent/src/integration_tests/graphql/queries/live_specs_filter_catalog_name.graphql new file mode 100644 index 00000000000..c31862c5c1d --- /dev/null +++ b/crates/agent/src/integration_tests/graphql/queries/live_specs_filter_catalog_name.graphql @@ -0,0 +1,35 @@ +# The composable `filter` narrows results to the caller's authorized specs. +# Alice reads all of aliceCo/, so `startsWith` returns her whole shared +# subtree and `in` matches both of its entries. Bob reads aliceCo/shared/ via +# a role grant, so he sees the same shared subtree but the `in` entry for +# Alice's private capture is dropped rather than erroring. +query LiveSpecsFilterCatalogName { + startsWith: liveSpecs( + filter: { catalogName: { startsWith: "aliceCo/shared/" } } + ) { + edges { + node { + catalogName + userCapability + } + } + } + inSet: liveSpecs( + filter: { + catalogName: { + in: [ + "aliceCo/shared/a" + "aliceCo/private/capture" + "bobCo/does-not-exist" + ] + } + } + ) { + edges { + node { + catalogName + userCapability + } + } + } +} diff --git a/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__alice-live_specs_filter_catalog_name.snap b/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__alice-live_specs_filter_catalog_name.snap new file mode 100644 index 00000000000..828c4a87c70 --- /dev/null +++ b/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__alice-live_specs_filter_catalog_name.snap @@ -0,0 +1,62 @@ +--- +source: crates/agent/src/integration_tests/graphql/queries/mod.rs +expression: alice_result +--- +{ + "inSet": { + "edges": [ + { + "node": { + "catalogName": "aliceCo/private/capture", + "userCapability": "admin" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/a", + "userCapability": "admin" + } + } + ] + }, + "startsWith": { + "edges": [ + { + "node": { + "catalogName": "aliceCo/shared/a", + "userCapability": "admin" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/b", + "userCapability": "admin" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/c", + "userCapability": "admin" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/capture", + "userCapability": "admin" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/disabled", + "userCapability": "admin" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/materialize", + "userCapability": "admin" + } + } + ] + } +} diff --git a/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__alice-live_specs_missing_name_and_prefix.snap b/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__alice-live_specs_missing_name_and_prefix.snap index ae266bc11a6..45369ac643a 100644 --- a/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__alice-live_specs_missing_name_and_prefix.snap +++ b/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__alice-live_specs_missing_name_and_prefix.snap @@ -2,4 +2,4 @@ source: crates/agent/src/integration_tests/graphql/queries/mod.rs expression: alice_result --- -"GraphQL query failed with error: GraphQL errors: [ServerError { message: \"must provide at least one of `names` or `prefix`\", locations: [Pos(2:5)], path: [Field(\"liveSpecs\")], extensions: None }]" +"GraphQL query failed with error: GraphQL errors: [ServerError { message: \"must provide at least one of `names` or `prefix`, or omit `by` entirely\", locations: [Pos(2:5)], path: [Field(\"liveSpecs\")], extensions: None }]" diff --git a/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__bob-live_specs_filter_catalog_name.snap b/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__bob-live_specs_filter_catalog_name.snap new file mode 100644 index 00000000000..61ad9443653 --- /dev/null +++ b/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__bob-live_specs_filter_catalog_name.snap @@ -0,0 +1,56 @@ +--- +source: crates/agent/src/integration_tests/graphql/queries/mod.rs +expression: bob_result +--- +{ + "inSet": { + "edges": [ + { + "node": { + "catalogName": "aliceCo/shared/a", + "userCapability": "read" + } + } + ] + }, + "startsWith": { + "edges": [ + { + "node": { + "catalogName": "aliceCo/shared/a", + "userCapability": "read" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/b", + "userCapability": "read" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/c", + "userCapability": "read" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/capture", + "userCapability": "read" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/disabled", + "userCapability": "read" + } + }, + { + "node": { + "catalogName": "aliceCo/shared/materialize", + "userCapability": "read" + } + } + ] + } +} diff --git a/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__bob-live_specs_missing_name_and_prefix.snap b/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__bob-live_specs_missing_name_and_prefix.snap index 206a5b78ac1..49c2ab09eb0 100644 --- a/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__bob-live_specs_missing_name_and_prefix.snap +++ b/crates/agent/src/integration_tests/graphql/queries/snapshots/agent__integration_tests__graphql__queries__bob-live_specs_missing_name_and_prefix.snap @@ -2,4 +2,4 @@ source: crates/agent/src/integration_tests/graphql/queries/mod.rs expression: bob_result --- -"GraphQL query failed with error: GraphQL errors: [ServerError { message: \"must provide at least one of `names` or `prefix`\", locations: [Pos(2:5)], path: [Field(\"liveSpecs\")], extensions: None }]" +"GraphQL query failed with error: GraphQL errors: [ServerError { message: \"must provide at least one of `names` or `prefix`, or omit `by` entirely\", locations: [Pos(2:5)], path: [Field(\"liveSpecs\")], extensions: None }]" diff --git a/crates/control-plane-api/src/server/public/graphql/live_spec_refs.rs b/crates/control-plane-api/src/server/public/graphql/live_spec_refs.rs index b594f96901f..fab6c095001 100644 --- a/crates/control-plane-api/src/server/public/graphql/live_spec_refs.rs +++ b/crates/control-plane-api/src/server/public/graphql/live_spec_refs.rs @@ -1,6 +1,8 @@ use crate::{ alerts::Alert, - server::public::graphql::{PgDataLoader, alerts, live_specs, publication_history, status}, + server::public::graphql::{ + PgDataLoader, alerts, filters, live_specs, publication_history, status, + }, }; use async_graphql::{ ComplexObject, Context, SimpleObject, dataloader, @@ -8,6 +10,7 @@ use async_graphql::{ }; const DEFAULT_PAGE_SIZE: usize = 50; +const MAX_PREFIXES: usize = 20; /// Input type for querying live specs. #[derive(Debug, Clone, async_graphql::InputObject)] @@ -22,6 +25,20 @@ pub struct LiveSpecsBy { pub data_plane_name: Option, } +/// Composable filter for the `liveSpecs` query. Every field is optional and +/// only narrows the result set; the caller's catalog-read scope is enforced +/// independently, so a filter can never widen what a caller may see. +#[derive(Debug, Clone, Default, async_graphql::InputObject)] +pub struct LiveSpecsFilter { + /// Narrow by catalog name. `startsWith` matches a whole subtree — specs + /// under `acmeCo/`, `acmeCo/team/`, etc. — like the deprecated + /// `by: { prefix }`. `in` matches an exact set of names, like + /// `by: { names }`. The two are alternative query modes and are mutually + /// exclusive. Either way, results compose with (never widen past) the + /// caller's authorized read prefixes. + pub catalog_name: Option, +} + /// Represents a reference from one live spec to another. #[derive(Debug, Clone, SimpleObject)] #[graphql(complex)] @@ -260,8 +277,13 @@ pub struct LiveSpecsQuery; #[async_graphql::Object] impl LiveSpecsQuery { - /// Returns a paginated list of live specs under the given prefix and - /// matching the given type. + /// Returns a paginated list of live specs accessible to the current user. + /// + /// Omitting both `by` and `filter` returns every live spec under every + /// prefix where the caller has catalog-read capability, and the optional + /// `filter` narrows those authorized results. The deprecated `by` instead + /// requires an explicit `names` or `prefix` selection, and fails the + /// entire request if any part of that selection is unauthorized. /// /// Note that the `user_capability` that's returned as part of the reference /// represents the user's capability to the whole prefix, and it is possible @@ -271,7 +293,12 @@ impl LiveSpecsQuery { pub async fn live_specs( &self, ctx: &Context<'_>, - by: LiveSpecsBy, + #[graphql( + deprecation = "Prefer `filter: { catalogName }`: `startsWith` replaces `prefix` and \ + `in` replaces `names`. `by` is retained only for existing clients." + )] + by: Option, + filter: Option, after: Option, before: Option, first: Option, @@ -279,29 +306,83 @@ impl LiveSpecsQuery { ) -> async_graphql::Result { let env = ctx.data::()?; - let LiveSpecsBy { - names, - prefix, - catalog_type, - data_plane_name: data_plane, - } = by; - let names = names.unwrap_or_default(); - - // Fail the entire request if it passed a name or prefix that the user is unauthorized to. - let policy_result = crate::server::evaluate_names_authorization( - env.snapshot(), - env.claims()?, - models::Capability::Read, - names - .iter() - .map(models::Name::as_str) - .chain(prefix.as_ref().map(models::Prefix::as_str).into_iter()), - ); - let (_expiry, ()) = env.authorization_outcome(policy_result).await?; + // `filter` is the going-forward replacement for `by`, and the two are + // mutually exclusive. They also authorize differently: `by` fails the + // entire request on an unauthorized name or prefix, while `filter` + // only narrows the caller's authorized prefixes and can never widen + // them. Both resolve into the parameters of one shared SQL query. + let filter_catalog_name = filter.and_then(|f| f.catalog_name); + let (names, prefix, catalog_type, data_plane, exact, read_prefixes) = match by { + Some(by) => { + if filter_catalog_name + .is_some_and(|cn| cn.starts_with.is_some() || cn.r#in.is_some()) + { + return Err( + "provide either `by` or `filter`, not both; `by` is deprecated".into(), + ); + } + let LiveSpecsBy { + names, + prefix, + catalog_type, + data_plane_name: data_plane, + } = by; + let names = names.unwrap_or_default(); - if names.is_empty() && prefix.is_none() { - return Err("must provide at least one of `names` or `prefix`".into()); - } + // Fail the entire request if it passed a name or prefix that the user is unauthorized to. + let policy_result = crate::server::evaluate_names_authorization( + env.snapshot(), + env.claims()?, + models::Capability::Read, + names + .iter() + .map(models::Name::as_str) + .chain(prefix.as_ref().map(models::Prefix::as_str).into_iter()), + ); + let (_expiry, ()) = env.authorization_outcome(policy_result).await?; + + if names.is_empty() && prefix.is_none() { + return Err( + "must provide at least one of `names` or `prefix`, or omit `by` entirely" + .into(), + ); + } + let prefix = prefix.map(|p| p.to_string()); + ( + names, + prefix, + catalog_type, + data_plane, + Vec::new(), + Vec::new(), + ) + } + None => { + let snapshot = env.snapshot(); + let (read_prefixes, starts_with, exact) = + super::authorized_prefixes::filtered_authorized_prefixes( + &snapshot.role_grants, + &snapshot.user_grants, + env.claims()?.sub, + models::authz::Capability::CatalogRead, + filter_catalog_name, + "filter.catalogName", + ) + .map(|(prefixes, starts_with, r#in)| { + (prefixes, starts_with, r#in.unwrap_or_default()) + })?; + + if read_prefixes.is_empty() { + return Ok(PaginatedLiveSpecsRefs::new(false, false)); + } + if read_prefixes.len() > MAX_PREFIXES { + return Err(async_graphql::Error::new( + "Too many accessible prefixes; narrow results with a filter", + )); + } + (Vec::new(), starts_with, None, None, exact, read_prefixes) + } + }; let (names, has_prev, has_next) = connection::query_with::( @@ -319,9 +400,11 @@ impl LiveSpecsQuery { let names = fetch_live_specs_names_before( &env.pg_pool, names, - prefix, + prefix.as_deref(), catalog_type, data_plane.as_deref(), + &exact, + &read_prefixes, before.as_deref(), limit as i64, ) @@ -336,9 +419,11 @@ impl LiveSpecsQuery { let names = fetch_live_specs_names_after( &env.pg_pool, names, - prefix, + prefix.as_deref(), catalog_type, data_plane.as_deref(), + &exact, + &read_prefixes, after.as_deref(), limit as i64, ) @@ -382,16 +467,21 @@ impl LiveSpecsQuery { async fn fetch_live_specs_names_after( db: &sqlx::PgPool, names: Vec, - prefix: Option, + prefix: Option<&str>, catalog_type: Option, data_plane: Option<&str>, + exact: &[String], + read_prefixes: &[String], after: Option<&str>, limit: i64, ) -> anyhow::Result> { assert!( - !names.is_empty() || prefix.is_some(), - "must have name or prefix predicate when querying live specs" + !names.is_empty() || prefix.is_some() || !read_prefixes.is_empty(), + "must have a name, prefix, or read-prefixes predicate when querying live specs" ); + // `exact` and `read_prefixes` are caller- and grant-derived strings, so + // they bind as text[] rather than the catalog_name domain array: sqlx does + // not run domain-constraint validation on bind. let names = sqlx::query_scalar!( r#"select ls.catalog_name as "name!: String" from live_specs ls @@ -400,13 +490,17 @@ async fn fetch_live_specs_names_after( and ($2::text is null or ls.catalog_name::text ^@ $2::text) and ($3::catalog_spec_type is null or ls.spec_type = $3::catalog_spec_type) and ($4::text is null or $4::text = dp.data_plane_name) - and ($5::catalog_name is null or ls.catalog_name > $5::catalog_name) + and (coalesce(array_length($5::text[], 1), 0) = 0 or ls.catalog_name::text = any($5::text[])) + and (coalesce(array_length($6::text[], 1), 0) = 0 or ls.catalog_name::text ^@ any($6::text[])) + and ($7::catalog_name is null or ls.catalog_name > $7::catalog_name) order by ls.catalog_name asc - limit $6"#, + limit $8"#, names as Vec, - prefix as Option, + prefix as Option<&str>, catalog_type as Option, data_plane as Option<&str>, + exact as &[String], + read_prefixes as &[String], after as Option<&str>, limit ) @@ -421,16 +515,21 @@ async fn fetch_live_specs_names_after( async fn fetch_live_specs_names_before( db: &sqlx::PgPool, names: Vec, - prefix: Option, + prefix: Option<&str>, catalog_type: Option, data_plane: Option<&str>, + exact: &[String], + read_prefixes: &[String], before: Option<&str>, limit: i64, ) -> anyhow::Result> { assert!( - !names.is_empty() || prefix.is_some(), - "must have name or prefix predicate when querying live specs" + !names.is_empty() || prefix.is_some() || !read_prefixes.is_empty(), + "must have a name, prefix, or read-prefixes predicate when querying live specs" ); + // `exact` and `read_prefixes` are caller- and grant-derived strings, so + // they bind as text[] rather than the catalog_name domain array: sqlx does + // not run domain-constraint validation on bind. let mut names = sqlx::query_scalar!( r#"select ls.catalog_name as "name!: String" from live_specs ls @@ -439,13 +538,17 @@ async fn fetch_live_specs_names_before( and ($2::text is null or ls.catalog_name::text ^@ $2::text) and ($3::catalog_spec_type is null or ls.spec_type = $3::catalog_spec_type) and ($4::text is null or $4::text = dp.data_plane_name) - and ($5::catalog_name is null or ls.catalog_name < $5::catalog_name) + and (coalesce(array_length($5::text[], 1), 0) = 0 or ls.catalog_name::text = any($5::text[])) + and (coalesce(array_length($6::text[], 1), 0) = 0 or ls.catalog_name::text ^@ any($6::text[])) + and ($7::catalog_name is null or ls.catalog_name < $7::catalog_name) order by ls.catalog_name desc - limit $6"#, + limit $8"#, names as Vec, - prefix as Option, + prefix as Option<&str>, catalog_type as Option, data_plane as Option<&str>, + exact as &[String], + read_prefixes as &[String], before as Option<&str>, limit ) @@ -455,3 +558,266 @@ async fn fetch_live_specs_names_before( names.reverse(); Ok(names) } + +#[cfg(test)] +mod test { + use crate::test_server; + + // Helper: run the query with the given variables and return the list of + // returned catalog names, asserting no GraphQL errors. + async fn query_names( + server: &test_server::TestServer, + token: &str, + variables: serde_json::Value, + ) -> Vec { + let response: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + query($by: LiveSpecsBy, $filter: LiveSpecsFilter) { + liveSpecs(by: $by, filter: $filter) { + edges { node { catalogName } } + } + } + "#, + "variables": variables, + }), + Some(token), + ) + .await; + assert!( + response.get("errors").is_none(), + "unexpected errors: {response}" + ); + response["data"]["liveSpecs"]["edges"] + .as_array() + .expect("edges array") + .iter() + .map(|edge| edge["node"]["catalogName"].as_str().unwrap().to_string()) + .collect() + } + + // Helper: assert a set of variables is rejected with a GraphQL error whose + // first message contains `expected_message`. + async fn expect_error( + server: &test_server::TestServer, + token: &str, + variables: serde_json::Value, + expected_message: &str, + ) { + let response: serde_json::Value = server + .graphql( + &serde_json::json!({ + "query": r#" + query($by: LiveSpecsBy, $filter: LiveSpecsFilter) { + liveSpecs(by: $by, filter: $filter) { + edges { node { catalogName } } + } + } + "#, + "variables": variables, + }), + Some(token), + ) + .await; + let message = response["errors"][0]["message"] + .as_str() + .unwrap_or_default(); + assert!( + message.contains(expected_message), + "expected error containing {expected_message:?}, got {response} for variables {variables}" + ); + } + + // The `filter` argument scopes results to the caller's authorized + // prefixes, narrowing by catalog-name subtree (`startsWith`) or exact set + // (`in`). The deprecated `by` keeps its stricter contract: it requires a + // `names` or `prefix` selection and fails outright on unauthorized names. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../../../fixtures", scripts("data_planes", "alice")) + )] + async fn live_specs_filter_scopes_by_catalog_name(pool: sqlx::PgPool) { + let _guard = test_server::init(); + + let snapshot = test_server::snapshot(pool.clone(), false).await; + let server = test_server::TestServer::start(pool.clone(), snapshot).await; + let alice_token = server.make_access_token(uuid::Uuid::from_bytes([0x11; 16]), None); + let bob_token = server.make_access_token(uuid::Uuid::from_bytes([0x22; 16]), None); + + // Omitting both `by` and `filter` returns every readable spec: alice + // reads aliceCo/ (admin) and ops/dp/public/ (role grant), and the ops + // fixture specs live outside that scope. + let no_filter = query_names(&server, &alice_token, serde_json::json!({})).await; + assert_eq!( + no_filter, + vec![ + "aliceCo/data/foo", + "aliceCo/in/capture-foo", + "aliceCo/out/materialize-bar" + ] + ); + + // A `startsWith` filter narrows to the matching subtree, the same + // result the deprecated `by: { prefix }` produces. + let narrowed = query_names( + &server, + &alice_token, + serde_json::json!({ "filter": { "catalogName": { "startsWith": "aliceCo/data/" } } }), + ) + .await; + assert_eq!(narrowed, vec!["aliceCo/data/foo"]); + + // The filter can never widen scope past the caller's grants. + let cross_tenant = query_names( + &server, + &alice_token, + serde_json::json!({ "filter": { "catalogName": { "startsWith": "ops/tasks/" } } }), + ) + .await; + assert!(cross_tenant.is_empty()); + + // An empty filter — and an empty `catalogName` within it — behave like + // omitting the filter: neither narrows anything. + let empty_filter = + query_names(&server, &alice_token, serde_json::json!({ "filter": {} })).await; + assert_eq!(empty_filter, no_filter); + let empty_catalog_name = query_names( + &server, + &alice_token, + serde_json::json!({ "filter": { "catalogName": {} } }), + ) + .await; + assert_eq!(empty_catalog_name, no_filter); + + // `in` matches an exact set of names, like `by: { names }`. Entries + // outside the caller's scope or naming nothing are dropped rather than + // erroring — unlike `by`, which fails the request on the ops entry. + let exact = query_names( + &server, + &alice_token, + serde_json::json!({ + "filter": { "catalogName": { "in": [ + "aliceCo/in/capture-foo", + "ops/tasks/public/one/logs", + "aliceCo/does-not-exist", + ] } } + }), + ) + .await; + assert_eq!(exact, vec!["aliceCo/in/capture-foo"]); + + // A caller with no grants sees an empty result, not an error. + let no_access = query_names(&server, &bob_token, serde_json::json!({})).await; + assert!(no_access.is_empty()); + + // `by` and `filter` are mutually exclusive. + expect_error( + &server, + &alice_token, + serde_json::json!({ + "by": { "prefix": "aliceCo/" }, + "filter": { "catalogName": { "startsWith": "aliceCo/" } }, + }), + "provide either `by` or `filter`", + ) + .await; + + // Within a filter, `startsWith` and `in` are mutually exclusive. + expect_error( + &server, + &alice_token, + serde_json::json!({ + "filter": { "catalogName": { + "startsWith": "aliceCo/", + "in": ["aliceCo/data/foo"], + } }, + }), + "mutually exclusive; provide only one", + ) + .await; + + // An empty `in` set is rejected at input validation, rather than + // ambiguously meaning "match nothing" or "match everything". + expect_error( + &server, + &alice_token, + serde_json::json!({ "filter": { "catalogName": { "in": [] } } }), + "", + ) + .await; + + // The deprecated `by` keeps requiring a `names` or `prefix` selection... + expect_error( + &server, + &alice_token, + serde_json::json!({ "by": { "catalogType": "capture" } }), + "must provide at least one of `names` or `prefix`, or omit `by` entirely", + ) + .await; + + // ...and keeps failing the entire request on an unauthorized prefix, + // where `filter` returns an empty result instead. + expect_error( + &server, + &alice_token, + serde_json::json!({ "by": { "prefix": "ops/tasks/" } }), + "not authorized", + ) + .await; + } + + // A caller who can read more than MAX_PREFIXES prefixes is refused an + // unfiltered listing, but succeeds once a filter narrows the authorized + // set back under the cap. + #[sqlx::test(migrations = "../../supabase/migrations")] + async fn live_specs_filter_narrows_below_max_prefixes(pool: sqlx::PgPool) { + let _guard = test_server::init(); + + let carol_id = uuid::Uuid::from_bytes([0x33; 16]); + sqlx::query("insert into auth.users (id, email) values ($1, 'carol@example.com')") + .bind(carol_id) + .execute(&pool) + .await + .unwrap(); + for i in 0..super::MAX_PREFIXES + 1 { + sqlx::query( + "insert into user_grants (user_id, object_role, capability) values ($1, $2, 'read')", + ) + .bind(carol_id) + .bind(format!("tenant{i:02}/")) + .execute(&pool) + .await + .unwrap(); + } + + let snapshot = test_server::snapshot(pool.clone(), false).await; + let server = test_server::TestServer::start(pool.clone(), snapshot).await; + let carol_token = server.make_access_token(carol_id, None); + + expect_error( + &server, + &carol_token, + serde_json::json!({}), + "Too many accessible prefixes", + ) + .await; + + // Both filter modes narrow the authorized set back under the cap. No + // specs exist under these prefixes, so the results are simply empty. + let narrowed = query_names( + &server, + &carol_token, + serde_json::json!({ "filter": { "catalogName": { "startsWith": "tenant00/" } } }), + ) + .await; + assert!(narrowed.is_empty()); + let narrowed = query_names( + &server, + &carol_token, + serde_json::json!({ "filter": { "catalogName": { "in": ["tenant01/some/spec"] } } }), + ) + .await; + assert!(narrowed.is_empty()); + } +} diff --git a/crates/flow-client/control-plane-api.graphql b/crates/flow-client/control-plane-api.graphql index 2444679a2e6..ee87179f3ac 100644 --- a/crates/flow-client/control-plane-api.graphql +++ b/crates/flow-client/control-plane-api.graphql @@ -1237,6 +1237,23 @@ input LiveSpecsBy { dataPlaneName: Name } +""" +Composable filter for the `liveSpecs` query. Every field is optional and +only narrows the result set; the caller's catalog-read scope is enforced +independently, so a filter can never widen what a caller may see. +""" +input LiveSpecsFilter { + """ + Narrow by catalog name. `startsWith` matches a whole subtree — specs + under `acmeCo/`, `acmeCo/team/`, etc. — like the deprecated + `by: { prefix }`. `in` matches an exact set of names, like + `by: { names }`. The two are alternative query modes and are mutually + exclusive. Either way, results compose with (never widen past) the + caller's authorized read prefixes. + """ + catalogName: PrefixFilter +} + """ Represents an optimistic lock failure when trying to update a specification. This typically means that an expectPubId was not matched, implying that @@ -1819,8 +1836,13 @@ type PublicationStatus { type QueryRoot { """ - Returns a paginated list of live specs under the given prefix and - matching the given type. + Returns a paginated list of live specs accessible to the current user. + + Omitting both `by` and `filter` returns every live spec under every + prefix where the caller has catalog-read capability, and the optional + `filter` narrows those authorized results. The deprecated `by` instead + requires an explicit `names` or `prefix` selection, and fails the + entire request if any part of that selection is unauthorized. Note that the `user_capability` that's returned as part of the reference represents the user's capability to the whole prefix, and it is possible @@ -1828,7 +1850,7 @@ type QueryRoot { words, this capability represents the _minimum_ capability that the user has for the given spec. """ - liveSpecs(by: LiveSpecsBy!, after: String, before: String, first: Int, last: Int): LiveSpecRefConnection! + liveSpecs(by: LiveSpecsBy @deprecated(reason: "Prefer `filter: { catalogName }`: `startsWith` replaces `prefix` and `in` replaces `names`. `by` is retained only for existing clients."), filter: LiveSpecsFilter, after: String, before: String, first: Int, last: Int): LiveSpecRefConnection! """ Returns a list of alerts that are currently active for the given catalog prefixes.