From 80a31820315d9dbd962eeeb328beb61b0bfdf7f1 Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Mon, 3 Aug 2026 15:43:22 +0000 Subject: [PATCH] control-plane: remove the remaining internal.user_roles() authorization call sites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follows #3155, which moved publication and discover authorization onto the in-memory Snapshot and left five internal.user_roles() call sites for follow-up. This removes all of them, by migration or by deletion. Evolutions: resolve_specs no longer filters by user_roles() in SQL. The query becomes a plain fetch (fetch_evolution_specs) and authorization moves to an in-process wrapper applying the shared spec-fetch policy: admin required per live spec, authoritative denials suppress the live join (preserving the "was never published" surfacing), and provisional denials surface as the retryable AuthorizationSnapshotStale error. Evolution gains a `started_at` anchor, threaded through its live-spec fetches in place of hardcoded None. One intentional delta, pinned by test: the Rust grant walk also traverses role grants whose subject is a parent of the held role, which the one-way SQL walk could not reach. The Rust implementation is authoritative over the legacy SQL semantics. Admin endpoints: create-data-plane and update-l2-reporting replace their raw ops/-admin SQL pre-check with the standard Envelope idiom (evaluate_names_authorization + authorization_outcome), extracted as server::authorize_ops_admin. The request's pinned Snapshot now serves both the pre-check and the ops publication: one request, one authorization view. A denial under a Snapshot which predates the request is now a 307 AuthZRetry rather than an immediate 403; the flow-plane-link unit gains `curl -L` so a redirect can't read as success. Storage-mappings directive: deleted rather than migrated. The GraphQL storage-mapping mutations superseded it on 2026-03-16 (the last routine production application), so the agent-side handler, its Directive variant, and user_has_admin_capability are removed — the latter taking with it a loose-direction prefix comparison that let a sub-prefix admin rewrite the whole tenant's mapping. The still-shared persistence helpers move out of directives/ to control-plane-api/src/storage_mappings.rs. That authorization boundary is now pinned on the GraphQL path instead: sub-prefix and attenuated raw-admin shapes are denied. The spec-fetch policy shared by get_live_specs, get_connected_live_specs, and evolutions::resolve_specs is centralized as Snapshot::spec_fetch_authorization so the three enforcement points cannot drift, and the per-module test Snapshot helpers consolidate into a shared test_support module. The agent crate now sets recursion_limit = 256: laying out the publication pipeline's nested poll future needs a ~130-deep rustc query descent, over the 128 default. Earlier trees compiled only because the since-deleted directive chain happened to pre-compute shared sub-layouts. Tests: evolutions parity/delta/staleness (sqlx), endpoint denial and 307-retry paths (test_server), GraphQL storage-mapping authorization denials, plus the pre-existing suites: 194/194 control-plane-api, 112/112 agent. --- ...e0acaca3d87b997e02da94513b584a2a0d6ce.json | 71 +++ ...bf45b2ea36fd6fb8be6e988627b2bdf45e21d.json | 22 - ...ccf86b5d2d08d19999930aabe4b7373d8c671.json | 72 --- ...c480c5ae585051ce828cae19ea96e25996aaa.json | 23 - crates/agent/src/directives/mod.rs | 18 +- .../agent/src/directives/storage_mappings.rs | 325 -------------- .../graphql/mutations/storage_mappings.rs | 63 +++ crates/agent/src/integration_tests/harness.rs | 2 +- crates/agent/src/lib.rs | 9 + crates/agent/src/main.rs | 2 +- .../control-plane-api/src/directives/mod.rs | 1 - crates/control-plane-api/src/evolutions/db.rs | 22 +- .../control-plane-api/src/evolutions/mod.rs | 422 +++++++++++++++++- crates/control-plane-api/src/lib.rs | 3 + .../control-plane-api/src/live_specs/mod.rs | 79 +--- .../src/publications/db_complete.rs | 21 +- .../src/publications/specs.rs | 45 +- .../src/server/create_data_plane.rs | 194 +++++++- crates/control-plane-api/src/server/mod.rs | 16 + .../server/public/graphql/storage_mappings.rs | 2 +- .../control-plane-api/src/server/snapshot.rs | 24 + .../src/server/update_l2_reporting.rs | 55 ++- .../src/{directives => }/storage_mappings.rs | 23 +- crates/control-plane-api/src/test_support.rs | 56 +++ local/systemd/flow-plane-link@.service | 11 + 25 files changed, 927 insertions(+), 654 deletions(-) create mode 100644 .sqlx/query-6fc77e0fec47ec83054f03aa4f0e0acaca3d87b997e02da94513b584a2a0d6ce.json delete mode 100644 .sqlx/query-752f293a6005958ed374bbf14bfbf45b2ea36fd6fb8be6e988627b2bdf45e21d.json delete mode 100644 .sqlx/query-bbb2f4e5602199ea3599771564cccf86b5d2d08d19999930aabe4b7373d8c671.json delete mode 100644 .sqlx/query-d87134fbea49426eb5f84ae7cdcc480c5ae585051ce828cae19ea96e25996aaa.json delete mode 100644 crates/agent/src/directives/storage_mappings.rs rename crates/control-plane-api/src/{directives => }/storage_mappings.rs (96%) create mode 100644 crates/control-plane-api/src/test_support.rs 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" +} 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/.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/.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..cd4a1c47f08 100644 --- a/crates/agent/src/directives/mod.rs +++ b/crates/agent/src/directives/mod.rs @@ -1,7 +1,4 @@ -use control_plane_api::{ - directives::{Row, fetch_directive, resolve}, - logs, -}; +use control_plane_api::directives::{Row, fetch_directive, resolve}; use anyhow::Context; use serde::{Deserialize, Serialize}; @@ -11,7 +8,6 @@ use validator::Validate; pub mod accept_demo_tenant; pub mod beta_onboard; pub mod click_to_accept; -pub mod storage_mappings; /// JobStatus is the possible outcomes of a handled directive operation. #[derive(Debug, Deserialize, Serialize)] @@ -36,26 +32,27 @@ impl JobStatus { } } +// The `storageMappings` directive type was removed after the GraphQL +// storage-mapping mutations superseded it (2026-03-16). A queued row of that +// type now resolves as `invalidDirective` (unknown variant) instead of being +// applied. #[derive(Debug, Deserialize, Serialize, schemars::JsonSchema)] #[serde(rename_all = "camelCase", tag = "type")] pub enum Directive { BetaOnboard(beta_onboard::Directive), ClickToAccept(click_to_accept::Directive), AcceptDemoTenant(accept_demo_tenant::Directive), - StorageMappings(storage_mappings::Directive), } #[derive(Clone)] pub struct DirectiveHandler { accounts_user_email: String, - logs_tx: logs::Tx, } impl DirectiveHandler { - pub fn new(accounts_user_email: String, logs_tx: &logs::Tx) -> Self { + pub fn new(accounts_user_email: String) -> Self { Self { accounts_user_email, - logs_tx: logs_tx.clone(), } } } @@ -132,9 +129,6 @@ 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? - } }; Ok(status) } diff --git a/crates/agent/src/directives/storage_mappings.rs b/crates/agent/src/directives/storage_mappings.rs deleted file mode 100644 index aece82ea866..00000000000 --- a/crates/agent/src/directives/storage_mappings.rs +++ /dev/null @@ -1,325 +0,0 @@ -use std::io::Write; - -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, - }, - }, - jobs, logs, -}; -use serde::{Deserialize, Serialize}; -use sqlx::types::Uuid; -use validator::Validate; - -#[derive(Debug, Deserialize, Serialize, Validate, schemars::JsonSchema)] -#[serde(rename_all = "camelCase", deny_unknown_fields)] -pub struct Directive {} - -#[derive(Debug, Deserialize, Serialize, Validate, schemars::JsonSchema)] -#[serde(rename_all = "camelCase", deny_unknown_fields)] -pub struct Claims { - add_store: models::Store, - catalog_prefix: models::Prefix, -} - -#[tracing::instrument(skip_all, ret, err, fields(row.user_claims))] -pub async fn apply( - _: Directive, - row: Row, - logs_tx: &logs::Tx, - txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, -) -> 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 { - Ok(c) => c, - Err(err) => { - return Ok(JobStatus::invalid_claims(err)); - } - }; - - let ProposedMapping { - catalog_prefix, - spec, - } = collection_data; - upsert_storage_mapping(Some(&detail), &catalog_prefix, spec, txn).await?; - let ProposedMapping { - catalog_prefix, - spec, - } = recovery; - upsert_storage_mapping(Some(&detail), &catalog_prefix, spec, txn).await?; - - Ok(JobStatus::Success) -} - -pub struct ProposedMapping { - catalog_prefix: String, - spec: models::StorageDef, -} - -fn add_store(stores: &mut models::StorageDef, store: models::Store) { - // If there's already an equivalent store, then remove it so that we don't end up with - // duplicates. This could happen if someone added store A, then store B, then store A again. - stores.stores.retain(|s| s != &store); - stores.stores.insert(0, store); -} - -async fn validate( - txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, - logs_tx: &logs::Tx, - row: Row, -) -> anyhow::Result<(ProposedMapping, ProposedMapping)> { - let claims: Claims = - serde_json::from_str(row.user_claims.get()).context("parsing user_claims")?; - - if !claims.catalog_prefix.ends_with('/') { - anyhow::bail!("invalid prefix, must end with '/'"); - } - - // Storage mappings can only be updated for an entire tenant. We may one day wish to support - // updates to narrower prefixes, but are trying to keep it simple for now. - if claims.catalog_prefix.matches('/').count() > 1 { - anyhow::bail!( - "catalog prefix contains too many path segments. Only top-level tenant prefixes can have storage mappings altered" - ); - } - - // 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 user_has_admin = - user_has_admin_capability(row.user_id, &claims.catalog_prefix, txn).await?; - anyhow::ensure!( - user_has_admin, - "user does not have required 'admin' capability to '{}'", - claims.catalog_prefix - ); - - // Check that we can actually access the storage bucket before fetching (and locking) the - // existing `storage_mappings` rows, since this check requires multiple network round trips. - check_bucket_access(row.logs_token, logs_tx, &claims.add_store).await?; - - let recovery_prefix = format!("recovery/{}", claims.catalog_prefix); - let existing_mappings = - fetch_storage_mappings(&claims.catalog_prefix, &recovery_prefix, txn).await?; - let (mut collection_data, mut recovery_data) = parse_existing(existing_mappings)?; - - let mut collection_store = claims.add_store.clone(); - // The storage mapping for collection data should always have a `collection-data/` prefix in - // order to segregate it from recovery log data. This makes it easier to apply bucket lifecycle - // policies, since you can target just the `collection-data/` prefix. If the store already - // specifies a prefix, then we'll add `collection-data/` to the end. - collection_store - .prefix_mut() - .as_mut_string() - .push_str("collection-data/"); - add_store(&mut collection_data.spec, collection_store); - // The recovery log store doesn't need a separate prefix because all recovery log journals - // already begin with `recovery/`. - add_store(&mut recovery_data.spec, claims.add_store.clone()); - - Ok((collection_data, recovery_data)) -} - -// 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 { - ($prog:expr, $($arg:expr),* $(; region $region:expr)?) => {{ - let mut cmd = std::process::Command::new($prog); - $( - cmd.arg($arg); - )* - $( - if let Some(region_name) = $region { - cmd.arg("--region"); - cmd.arg(region_name); - } - )? - cmd - } }; -} - -async fn check_bucket_access( - logs_token: Uuid, - logs_tx: &logs::Tx, - store: &models::Store, -) -> anyhow::Result<()> { - let mut test_file = tempfile::NamedTempFile::new().context("creating temp file")?; - test_file - .write_all(TEST_FILE_CONTENT.as_bytes()) - .context("writing test file content")?; - let test_file_path = test_file.path(); - - let commands = match store { - models::Store::S3(conf) => vec![ - ( - "put object", - check_command!( - "aws", - "s3", - "cp", - test_file_path, - without_query(conf.as_url()).join(TEST_FILENAME)?.to_string() - ; region conf.region.as_ref()), - ), - // List comes after put, so the prefix, if configured, is guaranteed to exist. - ( - "list bucket", - check_command!("aws", "s3", "ls", without_query(conf.as_url()).to_string() - ; region conf.region.as_ref()), - ), - ( - "get object", - // Copy to stdout to avoid needing to cleanup a temp file. - // Don't use /dev/null because the cli will exit non-zero even when it gets the file successfully. - check_command!( - "aws", - "s3", - "cp", - without_query(conf.as_url()) - .join(TEST_FILENAME)? - .to_string(), - "/dev/stdout" - ; region conf.region.as_ref() - ), - ), - ( - "delete object", - check_command!( - "aws", - "s3", - "rm", - without_query(conf.as_url()).join(TEST_FILENAME)?.to_string() - ; region conf.region.as_ref() - ), - ), - ], - models::Store::Gcs(conf) => vec![ - ( - "put object", - check_command!( - "gcloud", - "storage", - "cp", - test_file_path, - without_query(conf.as_url()) - .join(TEST_FILENAME)? - .to_string() - ), - ), - // List comes after put, so the prefix, if configured, is guaranteed to exist. - ( - "list bucket", - check_command!( - "gcloud", - "storage", - "ls", - without_query(conf.as_url()).to_string() - ), - ), - ( - "get object", - check_command!( - "gcloud", - "storage", - "cat", - without_query(conf.as_url()) - .join(TEST_FILENAME)? - .to_string() - ), - ), - ( - "delete object", - check_command!( - "gcloud", - "storage", - "rm", - without_query(conf.as_url()) - .join(TEST_FILENAME)? - .to_string() - ), - ), - ], - models::Store::Azure(_) => { - anyhow::bail!("checking access for azure cloud storage is not yet implemented") - } - models::Store::Custom(_) => { - anyhow::bail!("checking access for custom cloud storage is not supported") - } - }; - - for (desc, mut cmd) in commands { - tracing::info!( - %desc, - program = ?cmd.get_program(), - args = ?cmd.get_args(), - "running storage check" - ); - let exit_status = jobs::run_without_removing_env(&desc, logs_tx, logs_token, &mut cmd) - .await - .with_context(|| { - format!( - "failed to execute {desc} command: {:?} with: {:?}", - cmd.get_program(), - cmd.get_args() - ) - })?; - if !exit_status.success() { - anyhow::bail!("failed to {desc}, please check that permissions are set appropriately"); - } - } - - Ok(()) -} - -fn without_query(mut uri: url::Url) -> url::Url { - uri.set_query(None); - uri -} - -fn parse_existing( - mut existing: Vec, -) -> anyhow::Result<(ProposedMapping, ProposedMapping)> { - if existing.len() != 2 { - anyhow::bail!("expected 2 existing storage mappings, found: {existing:?}"); - } - let Some(recovery_idx) = existing - .iter() - .position(|m| m.catalog_prefix.starts_with("recovery/")) - else { - anyhow::bail!("missing recovery/ storage mapping in {existing:?}"); - }; - let recovery = existing.remove(recovery_idx); - let recovery_storage: models::StorageDef = serde_json::from_str(recovery.spec.get()) - .context("deserializing existing recovery/ storage mapping")?; - - let collection_data = existing.remove(0); - let collection_store: models::StorageDef = serde_json::from_str(collection_data.spec.get()) - .context("deserializing existing storage mapping")?; - - Ok(( - ProposedMapping { - catalog_prefix: collection_data.catalog_prefix, - spec: collection_store, - }, - ProposedMapping { - catalog_prefix: recovery.catalog_prefix, - spec: recovery_storage, - }, - )) -} - -const TEST_FILENAME: &str = "estuary_test.txt"; - -const TEST_FILE_CONTENT: &str = r#"Estuary storage test -This file is written to your storage bucket in order to test that we have the necessary access -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."#; diff --git a/crates/agent/src/integration_tests/graphql/mutations/storage_mappings.rs b/crates/agent/src/integration_tests/graphql/mutations/storage_mappings.rs index 6cb6a624d99..f9311b3c24c 100644 --- a/crates/agent/src/integration_tests/graphql/mutations/storage_mappings.rs +++ b/crates/agent/src/integration_tests/graphql/mutations/storage_mappings.rs @@ -9,6 +9,69 @@ mutation CreateStorageMapping($catalogPrefix: Prefix!, $spec: JSON!) { } "#; +/// Storage-mapping mutations require effective (attenuation-aware) admin over +/// the *entire* claimed prefix, because a mapping at any prefix shadows the +/// tenant mapping for everything under it (longest-match wins). This pins the +/// authorization boundary formerly guarded by the removed `storageMappings` +/// applied-directive, whose legacy `internal.user_roles()` SQL accepted the +/// sub-prefix shape (it compared prefixes in the loose direction) and was +/// blind to bundle attenuation. +#[tokio::test] +async fn test_create_storage_mapping_authorization_denials() { + let mut harness = TestHarness::init("storage_mapping_authz").await; + let _alice = harness.setup_tenant("aliceCo").await; + + // bob administers only a *sub*-prefix: authority flows downward, so it + // must not reach the tenant root. + let bob = uuid::uuid!("bbbbbbbb-0000-0000-0000-000000000000"); + // erin reaches `aliceCo/` through a raw-`admin` role grant, but her own + // grant delegates only the `editor` bundle: the walk attenuates the + // second hop's bits below Admin. + let erin = uuid::uuid!("eeeeeeee-0000-0000-0000-000000000000"); + sqlx::query( + r#"with users as ( + insert into auth.users (id, email) values + ($1, 'subprefix-admin@example.test'), ($2, 'attenuated-admin@example.test') + ), + user_grants as ( + insert into user_grants (user_id, object_role, capability, bundles) values + ($1, 'aliceCo/sub/', 'admin', '{}'), + ($2, 'sharedCo/', 'none', '{editor}') + ) + insert into role_grants (subject_role, object_role, capability) values + ('sharedCo/', 'aliceCo/', 'admin')"#, + ) + .bind(bob) + .bind(erin) + .execute(&harness.pool) + .await + .unwrap(); + // An authoritative Snapshot (taken after the request starts) makes the + // denials terminal rather than retryable. + harness.refresh_snapshot_authoritative().await; + + for user_id in [bob, erin] { + let result: Result = harness + .execute_graphql_query( + user_id, + CREATE_STORAGE_MAPPING_MUTATION, + &json!({ + "catalogPrefix": "aliceCo/", + "spec": { + "stores": [{"provider": "GCS", "bucket": "test-bucket"}], + "data_planes": ["ops/dp/public/test"] + }, + }), + ) + .await; + let err = result.unwrap_err().to_string(); + assert!( + err.contains("is not an authorized as an Admin of catalog prefix 'aliceCo/'"), + "expected an admin denial for {user_id}, got: {err}" + ); + } +} + #[tokio::test] async fn test_create_storage_mapping_validation_errors() { let mut harness = TestHarness::init("storage_mapping_validation").await; diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index c6e3e2de1db..eb443c2e348 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -304,7 +304,7 @@ 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); + crate::directives::DirectiveHandler::new("support@estuary.test".to_string()); let mut harness = TestHarness { test_name, diff --git a/crates/agent/src/lib.rs b/crates/agent/src/lib.rs index 7d1c8127209..65955377628 100644 --- a/crates/agent/src/lib.rs +++ b/crates/agent/src/lib.rs @@ -1,3 +1,12 @@ +// Computing the layout of `PublicationsExecutor::poll`'s async body recurses +// through every future of the nested publication pipeline — a ~130-deep query +// descent, over rustc's default limit of 128. Earlier trees compiled only +// because since-deleted code (the storage-mappings directive chain) happened +// to pre-compute shared sub-layouts, splitting the descent below the limit. +// Raise the limit per rustc's own guidance rather than depend on that +// accident. +#![recursion_limit = "256"] + pub mod alerts; pub(crate) mod connector_tags; pub mod controllers; diff --git a/crates/agent/src/main.rs b/crates/agent/src/main.rs index 668d6f2eb8b..65a61140ee5 100644 --- a/crates/agent/src/main.rs +++ b/crates/agent/src/main.rs @@ -408,7 +408,7 @@ 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); 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/mod.rs b/crates/control-plane-api/src/directives/mod.rs index 1ed74a587b9..bb1a272a11e 100644 --- a/crates/control-plane-api/src/directives/mod.rs +++ b/crates/control-plane-api/src/directives/mod.rs @@ -1,6 +1,5 @@ pub mod accept_demo_tenant; pub mod beta_onboard; -pub mod storage_mappings; use crate::TextJson; use chrono::{DateTime, Utc}; diff --git a/crates/control-plane-api/src/evolutions/db.rs b/crates/control-plane-api/src/evolutions/db.rs index 5a641d6f5b9..cd18b6f773b 100644 --- a/crates/control-plane-api/src/evolutions/db.rs +++ b/crates/control-plane-api/src/evolutions/db.rs @@ -58,8 +58,9 @@ pub struct SpecRow { pub catalog_name: String, /// The id of the draft spec, or None if it is not already in the draft pub draft_spec_id: Option, - /// The id of the live spec, or None if the spec was never published (which - /// will be surfaced as an error) + /// The id of the live spec. None if the spec was never published (which + /// will be surfaced as an error), or if `evolutions::resolve_specs` + /// suppressed the live side because the user isn't authorized to it. pub live_spec_id: Option, /// The current value of `expect_pub_id` from the draft spec, if drafted pub expect_pub_id: Option, @@ -78,8 +79,10 @@ pub struct SpecRow { /// up affecting them, but we cannot know for certain until we check all of their /// 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, +/// +/// This is a plain fetch: user authorization is applied in-process by +/// `evolutions::resolve_specs` against a Snapshot, not in SQL. +pub async fn fetch_evolution_specs( draft_id: Id, collection_names: Vec, txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, @@ -99,12 +102,10 @@ 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 internal.user_roles($2, 'admin') 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 ), @@ -117,9 +118,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 internal.user_roles($2, 'admin') r where ls.catalog_name ^@ r.role_prefix) ) select catalog_name as "catalog_name!: String", @@ -142,9 +140,9 @@ pub async fn resolve_specs( from live "#, draft_id as Id, - user_id as Uuid, collection_names as Vec, - ).fetch_all(&mut **txn) + ) + .fetch_all(&mut **txn) .await } diff --git a/crates/control-plane-api/src/evolutions/mod.rs b/crates/control-plane-api/src/evolutions/mod.rs index aba5f647dff..edd732af74f 100644 --- a/crates/control-plane-api/src/evolutions/mod.rs +++ b/crates/control-plane-api/src/evolutions/mod.rs @@ -1,7 +1,7 @@ mod db; use crate::Snapshot; -pub use db::{Row, fetch_evolution, fetch_resource_spec_schema, resolve, resolve_specs}; +pub use db::{Row, SpecRow, fetch_evolution, fetch_resource_spec_schema, resolve}; use itertools::Itertools; pub use models::{Capability, evolutions::EvolvedCollection}; use serde::{Deserialize, Serialize}; @@ -24,6 +24,12 @@ pub struct Evolution { /// and `false` for evolutions that are undertaken by our background /// automations. pub require_user_can_admin: bool, + /// The instant the evolution was queued (the `evolutions` row's + /// `updated_at`), which anchors authorization staleness: a Snapshot denial + /// is authoritative only once the Snapshot postdates it. `None` — for + /// callers without a durable queued instant — falls back to anchoring each + /// denied spec on its own last publication time. + pub started_at: Option, } #[derive(Debug)] @@ -122,6 +128,53 @@ impl EvolveRequest { } } +/// Fetches the specs needed by an evolutions job and applies user +/// authorization in-process against `snapshot`, replacing the recursive +/// `internal.user_roles()` filtering the fetch formerly did in SQL. +/// +/// The user must hold `admin` to affect a live spec. A drafted spec whose +/// live counterpart is denied keeps its drafted side but loses the live join +/// (surfacing downstream as "was never published", the pre-existing +/// behavior); a denied not-drafted live spec is dropped entirely. As with +/// `live_specs::get_live_specs`, a denial is trusted only once the Snapshot +/// is authoritative for `started_at` — or, absent one, for the denied spec's +/// own last publication — and otherwise surfaces as a retryable +/// `AuthorizationSnapshotStale` error; cancelling the Snapshot's `revoke` to +/// request an early refresh is the caller's responsibility, as in `evolve`. +pub async fn resolve_specs( + user_id: uuid::Uuid, + draft_id: models::Id, + collection_names: Vec, + txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, + snapshot: &Snapshot, + started_at: Option, +) -> anyhow::Result> { + let rows = db::fetch_evolution_specs(draft_id, collection_names, txn).await?; + + let mut out = Vec::with_capacity(rows.len()); + for mut row in rows { + if let Some(last_pub_id) = row.last_pub_id { + let authorized = snapshot.spec_fetch_authorization( + user_id, + &row.catalog_name, + Capability::Admin, + started_at, + last_pub_id, + )?; + + if !authorized { + if row.draft_spec_id.is_none() { + continue; + } + row.live_spec_id = None; + row.last_pub_id = None; + } + } + out.push(row); + } + Ok(out) +} + #[tracing::instrument(skip_all, fields(user_id = %evolution.user_id))] pub async fn evolve( evolution: Evolution, @@ -133,6 +186,7 @@ pub async fn evolve( requests, user_id, require_user_can_admin, + started_at, } = evolution; for req in requests.iter() { if let Err(error) = req.validate() { @@ -174,7 +228,7 @@ pub async fn evolve( capability_filter, db, snapshot, - None, + started_at, ) .await { @@ -203,9 +257,7 @@ pub async fn evolve( capability_filter, db, snapshot, - // `evolve` is not handed the queued `evolutions` row, so it has no - // durable instant to anchor staleness on and falls back to per-spec. - None, + started_at, ) .await { @@ -450,6 +502,366 @@ lazy_static::lazy_static! { static ref NAME_VERSION_RE: regex::Regex = regex::Regex::new(r#".*[_-][vV](\d+)$"#).unwrap(); } +/// These tests pin the privilege boundary of `resolve_specs` across its +/// migration from in-SQL `internal.user_roles()` filtering to Snapshot-based +/// authorization: nobody gains or loses authority relative to the legacy SQL, +/// except deltas inherent to the Rust grant walk +/// (`tables::UserGrant::is_authorized`), which is authoritative over the +/// legacy semantics. The test marked ACCEPTED DELTA pins such an intentional +/// difference; the rest are parity cases that held under both models. +#[cfg(test)] +mod tests { + use super::*; + use crate::test_support::{assert_stale_for, authoritative, published_at, stale}; + + // From `fixtures/authz_specs.sql`. Carol is admin of `carolCo/`; Dan + // administers only `danCo/` and so models an unauthorized caller. + const CAROL: uuid::Uuid = uuid::uuid!("33333333-3333-3333-3333-333333333333"); + const DAN: uuid::Uuid = uuid::uuid!("44444444-4444-4444-4444-444444444444"); + const COLLECTION: &str = "carolCo/data/foo"; + const CAPTURE: &str = "carolCo/in/capture-foo"; + + const DRAFT_ID: &str = "11:11:11:11:11:11:11:11"; + + /// Inserts a draft owned by `user_id` which drafts each of `names`, and + /// returns its id. Ids are fixed: tests run in isolated databases. + async fn insert_draft(pool: &sqlx::PgPool, user_id: uuid::Uuid, names: &[&str]) -> models::Id { + let draft_id: models::Id = sqlx::query_scalar( + "insert into drafts (id, user_id) values ($1::flowid, $2) returning id", + ) + .bind(DRAFT_ID) + .bind(user_id) + .fetch_one(pool) + .await + .expect("inserting draft"); + + for (index, name) in names.iter().enumerate() { + sqlx::query( + r#"insert into draft_specs (id, draft_id, catalog_name, spec, spec_type) + values ($1::flowid, $2::flowid, $3, '{}', 'collection')"#, + ) + .bind(format!("22:22:22:22:22:22:22:{:02x}", index)) + .bind(DRAFT_ID) + .bind(name) + .execute(pool) + .await + .expect("inserting draft spec"); + } + draft_id + } + + /// Adds a user holding the given grants, mirroring shapes from + /// `fixtures/attenuated_grants.sql`. `user_grant` is + /// (object_role, capability, bundles-literal like `'{editor}'` or `'{}'`). + async fn insert_user( + pool: &sqlx::PgPool, + user_id: uuid::Uuid, + email: &str, + user_grant: (&str, &str, &str), + role_grants: &[(&str, &str, &str)], + ) { + sqlx::query("insert into auth.users (id, email) values ($1, $2)") + .bind(user_id) + .bind(email) + .execute(pool) + .await + .expect("inserting user"); + + let (object_role, capability, bundles) = user_grant; + sqlx::query( + r#"insert into user_grants (user_id, object_role, capability, bundles) + values ($1, $2, $3::grant_capability, $4::capability_bundle[])"#, + ) + .bind(user_id) + .bind(object_role) + .bind(capability) + .bind(bundles) + .execute(pool) + .await + .expect("inserting user grant"); + + for (subject_role, object_role, capability) in role_grants { + sqlx::query( + r#"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 + .expect("inserting role grant"); + } + } + + /// Runs `resolve_specs` and returns rows sorted by catalog name. + async fn resolve_sorted( + pool: &sqlx::PgPool, + user_id: uuid::Uuid, + draft_id: models::Id, + collection_names: &[&str], + snapshot: &crate::Snapshot, + started_at: Option, + ) -> anyhow::Result> { + let mut txn = pool.begin().await.expect("begin"); + let mut rows = resolve_specs( + user_id, + draft_id, + collection_names.iter().map(|n| n.to_string()).collect(), + &mut txn, + snapshot, + started_at, + ) + .await?; + rows.sort_by(|l, r| l.catalog_name.cmp(&r.catalog_name)); + Ok(rows) + } + + /// A user holding admin directly on `carolCo/` resolves both the drafted + /// spec's live join and referenced not-drafted live specs. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_resolve_specs_admin_direct_grant(pool: sqlx::PgPool) { + let draft_id = insert_draft(&pool, CAROL, &[COLLECTION]).await; + // An authorized caller resolves identically under stale and + // authoritative Snapshots: staleness only converts denials. + for snapshot in [stale(&pool).await, authoritative(&pool).await] { + let rows = resolve_sorted(&pool, CAROL, draft_id, &[CAPTURE], &snapshot, None) + .await + .expect("carol is admin of carolCo/"); + + assert_eq!(2, rows.len(), "{rows:?}"); + assert_eq!(COLLECTION, rows[0].catalog_name); + assert!(rows[0].draft_spec_id.is_some()); + assert!(rows[0].live_spec_id.is_some(), "live join populated"); + assert!(rows[0].last_pub_id.is_some()); + assert_eq!(CAPTURE, rows[1].catalog_name); + assert!(rows[1].draft_spec_id.is_none()); + assert!(rows[1].live_spec_id.is_some()); + } + } + + /// Admin reached through a transitive role grant (eve → eveCo/ → carolCo/) + /// is equivalent to a direct grant, before and after the migration. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_resolve_specs_admin_via_transitive_role_grant(pool: sqlx::PgPool) { + let eve = uuid::uuid!("77777777-7777-7777-7777-777777777777"); + insert_user( + &pool, + eve, + "eve@example.com", + ("eveCo/", "admin", "{}"), + &[("eveCo/", "carolCo/", "admin")], + ) + .await; + + let draft_id = insert_draft(&pool, eve, &[COLLECTION]).await; + let rows = resolve_sorted( + &pool, + eve, + draft_id, + &[CAPTURE], + &authoritative(&pool).await, + None, + ) + .await + .expect("eve is admin of carolCo/ transitively"); + + assert_eq!(2, rows.len(), "{rows:?}"); + assert!(rows[0].live_spec_id.is_some(), "live join populated"); + assert_eq!(CAPTURE, rows[1].catalog_name); + assert!(rows[1].live_spec_id.is_some()); + } + + /// An unauthorized user's drafted spec loses its live join (surfacing + /// later as "was never published"), and referenced not-drafted live specs + /// are dropped entirely. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_resolve_specs_unauthorized_live_suppressed(pool: sqlx::PgPool) { + let draft_id = insert_draft(&pool, DAN, &[COLLECTION]).await; + let rows = resolve_sorted( + &pool, + DAN, + draft_id, + &[CAPTURE], + &authoritative(&pool).await, + None, + ) + .await + .expect("an authoritative denial is a silent suppression, not an error"); + + assert_eq!(1, rows.len(), "capture must be dropped: {rows:?}"); + assert_eq!(COLLECTION, rows[0].catalog_name); + assert!(rows[0].draft_spec_id.is_some()); + assert!(rows[0].live_spec_id.is_none(), "live join suppressed"); + assert!(rows[0].last_pub_id.is_none()); + } + + /// Drafted specs are always returned regardless of authorization; only + /// their live joins are subject to it. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_resolve_specs_drafted_returned_regardless(pool: sqlx::PgPool) { + let draft_id = insert_draft(&pool, DAN, &[COLLECTION, "danCo/new"]).await; + let rows = resolve_sorted(&pool, DAN, draft_id, &[], &authoritative(&pool).await, None) + .await + .expect("drafted specs resolve regardless of authorization"); + + assert_eq!(2, rows.len(), "{rows:?}"); + assert_eq!(COLLECTION, rows[0].catalog_name); + assert!(rows[0].live_spec_id.is_none()); + assert_eq!("danCo/new", rows[1].catalog_name); + assert!(rows[1].live_spec_id.is_none()); + } + + /// ACCEPTED DELTA — `internal.user_roles()` walked role_grants in a single + /// direction (subject starts-with the held role), so admin held on + /// `teamCo/nested/` could not use the `teamCo/ → carolCo/` grant and this + /// shape was denied. The Rust walk (`tables::UserGrant::is_authorized`) + /// also traverses grants whose subject is a *prefix* of the held role and + /// authorizes it — an intentional widening; the Rust implementation is + /// authoritative over the legacy SQL semantics (#control-plane, + /// 2026-04-13). + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_resolve_specs_parent_subject_role_grant(pool: sqlx::PgPool) { + let gwen = uuid::uuid!("88888888-8888-8888-8888-888888888888"); + insert_user( + &pool, + gwen, + "gwen@example.com", + ("teamCo/nested/", "admin", "{}"), + &[("teamCo/", "carolCo/", "admin")], + ) + .await; + + let draft_id = insert_draft(&pool, gwen, &[COLLECTION]).await; + let rows = resolve_sorted( + &pool, + gwen, + draft_id, + &[CAPTURE], + &authoritative(&pool).await, + None, + ) + .await + .expect("gwen reaches carolCo/ through the parent-subject grant"); + + assert_eq!(2, rows.len(), "{rows:?}"); + assert!(rows[0].live_spec_id.is_some(), "the Rust walk authorizes"); + assert_eq!(CAPTURE, rows[1].catalog_name); + assert!(rows[1].live_spec_id.is_some()); + } + + /// PARITY — an attenuated path (raw `none` capability delegating only the + /// `editor` bundle, then a raw-`admin` role grant) is denied under both + /// models: `user_roles('admin')` rejects the first hop's capability, and + /// the Rust walk attenuates the second hop's bits down to `editor`, which + /// does not satisfy `Admin`. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_resolve_specs_attenuated_admin_denied(pool: sqlx::PgPool) { + let erin = uuid::uuid!("55555555-5555-5555-5555-555555555555"); + insert_user( + &pool, + erin, + "erin@example.com", + ("sharedCo/", "none", "{editor}"), + &[("sharedCo/", "carolCo/", "admin")], + ) + .await; + + let draft_id = insert_draft(&pool, erin, &[COLLECTION]).await; + let rows = resolve_sorted( + &pool, + erin, + draft_id, + &[CAPTURE], + &authoritative(&pool).await, + None, + ) + .await + .expect("an authoritative denial is a silent suppression, not an error"); + + assert_eq!(1, rows.len(), "capture must be dropped: {rows:?}"); + assert!(rows[0].live_spec_id.is_none(), "live join suppressed"); + } + + /// A denial from a Snapshot which predates the denied spec's publication + /// is provisional: it surfaces as a retryable `AuthorizationSnapshotStale` + /// naming the spec, never as a silent suppression. Requesting an early + /// refresh (`snapshot.revoke`) is the calling executor's job, as in + /// `evolve`'s stale arms. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_resolve_specs_stale_denial_is_retryable(pool: sqlx::PgPool) { + let draft_id = insert_draft(&pool, DAN, &[COLLECTION]).await; + let err = resolve_sorted(&pool, DAN, draft_id, &[CAPTURE], &stale(&pool).await, None) + .await + .expect_err("a stale denial must be surfaced, not suppressed"); + assert_stale_for(err, COLLECTION); + } + + /// `started_at` displaces the per-spec staleness anchor in both + /// directions: a Snapshot which postdates the spec can still be stale for + /// a later-queued evolution, and a Snapshot which predates the spec is + /// authoritative for an earlier-queued one. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_resolve_specs_request_relative_staleness(pool: sqlx::PgPool) { + let draft_id = insert_draft(&pool, DAN, &[COLLECTION]).await; + let published = published_at(&pool).await; + let skew = crate::Snapshot::TEMPORAL_SKEW; + + // Snapshot postdates the spec (authoritative per-spec) but predates + // the queued evolution: the denial is provisional. + let err = resolve_sorted( + &pool, + DAN, + draft_id, + &[CAPTURE], + &authoritative(&pool).await, + Some(published + skew * 8), + ) + .await + .expect_err("denial under a Snapshot older than the request is provisional"); + assert_stale_for(err, COLLECTION); + + // Snapshot predates the spec (stale per-spec) but postdates the queued + // evolution: the denial is authoritative and silently suppresses. + let rows = resolve_sorted( + &pool, + DAN, + draft_id, + &[CAPTURE], + &stale(&pool).await, + Some(published - skew * 8), + ) + .await + .expect("denial under a Snapshot newer than the request is authoritative"); + assert_eq!(1, rows.len(), "capture must be dropped: {rows:?}"); + assert!(rows[0].live_spec_id.is_none(), "live join suppressed"); + } +} + /// Takes an existing name and returns a new name with an incremeted version suffix. /// The name `foo` will become `foo_v2`, and `foo_v2` will become `foo_v3` and so on. fn next_name(current_name: &str) -> String { diff --git a/crates/control-plane-api/src/lib.rs b/crates/control-plane-api/src/lib.rs index 2bf57e7a9ec..742c6d543d6 100644 --- a/crates/control-plane-api/src/lib.rs +++ b/crates/control-plane-api/src/lib.rs @@ -21,10 +21,13 @@ pub mod logs; pub mod proxy_connectors; pub mod publications; pub mod server; +pub mod storage_mappings; mod text_json; #[cfg(test)] pub(crate) mod test_server; +#[cfg(test)] +pub(crate) mod test_support; /// TextJson encodes JSON for Postgres while preserving property ordering. pub use text_json::TextJson; diff --git a/crates/control-plane-api/src/live_specs/mod.rs b/crates/control-plane-api/src/live_specs/mod.rs index 26d45b66fc0..6af7dd0ce6b 100644 --- a/crates/control-plane-api/src/live_specs/mod.rs +++ b/crates/control-plane-api/src/live_specs/mod.rs @@ -49,15 +49,13 @@ pub async fn get_live_specs( continue; }; if let Some(min_capability) = filter_capability { - // For discovers, anchor to the discover request time (started_at). - // For other callers, anchor to the spec's publication time. - // An authoritative denial is today's silent drop; a provisional - // one surfaces as a retryable stale error. - let anchor = started_at.unwrap_or_else(|| row.last_pub_id.timestamp()); - if !snapshot - .user_authorization(user_id, &row.catalog_name, min_capability, Some(anchor)) - .ok_or_stale(&row.catalog_name)? - { + if !snapshot.spec_fetch_authorization( + user_id, + &row.catalog_name, + min_capability, + started_at, + row.last_pub_id, + )? { continue; } } @@ -98,15 +96,13 @@ pub async fn get_connected_live_specs( for exp in expanded_rows { if let Some(minimum_capability) = filter_capability { - // Callers without a durable request time — those which capture - // "now" anew on every attempt and retry on their own — anchor to - // the spec's last publication, which bounds the window in which - // grants could have been committed alongside the spec itself. - let anchor = started.unwrap_or_else(|| exp.last_pub_id.timestamp()); - if !snapshot - .user_authorization(user_id, &exp.catalog_name, minimum_capability, Some(anchor)) - .ok_or_stale(&exp.catalog_name)? - { + if !snapshot.spec_fetch_authorization( + user_id, + &exp.catalog_name, + minimum_capability, + started, + exp.last_pub_id, + )? { continue; } } @@ -150,6 +146,7 @@ pub async fn get_connected_live_specs( #[cfg(test)] mod tests { use super::*; + use crate::test_support::{assert_stale_for, authoritative, published_at, stale}; // From `fixtures/authz_specs.sql`. Carol is admin of `carolCo/`; Dan holds no // grants at all and so models an unauthorized caller. @@ -158,52 +155,6 @@ mod tests { const COLLECTION: &str = "carolCo/data/foo"; const CAPTURE: &str = "carolCo/in/capture-foo"; - /// Staleness compares the Snapshot's `taken` against the timestamp embedded - /// in a spec's `last_pub_id`, so read that back rather than recomputing it — - /// `flowid` is `macaddr8`, which silently widens short literals. - async fn published_at(pool: &sqlx::PgPool) -> tokens::DateTime { - sqlx::query_scalar!( - r#"select last_pub_id as "last_pub_id: models::Id" - from live_specs where catalog_name = $1"#, - COLLECTION, - ) - .fetch_one(pool) - .await - .expect("fixture collection should exist") - .timestamp() - } - - /// A Snapshot holding the fixture's real grants, stamped `offset` away from - /// the instant the fixture's specs were published. - async fn snapshot_offset(pool: &sqlx::PgPool, offset: chrono::TimeDelta) -> crate::Snapshot { - let mut decrypted_hmac_keys = std::collections::HashMap::new(); - let data = crate::snapshot::try_fetch(pool, &mut decrypted_hmac_keys) - .await - .expect("failed to fetch snapshot"); - crate::Snapshot::new(published_at(pool).await + offset, data) - } - - /// Taken clear of the publication plus `TEMPORAL_SKEW`: denials are definitive. - async fn authoritative(pool: &sqlx::PgPool) -> crate::Snapshot { - snapshot_offset(pool, crate::Snapshot::TEMPORAL_SKEW * 4).await - } - - /// Taken before the publication it would judge: denials are retryable. - async fn stale(pool: &sqlx::PgPool) -> crate::Snapshot { - snapshot_offset(pool, -crate::Snapshot::TEMPORAL_SKEW * 4).await - } - - fn assert_stale_for(err: anyhow::Error, catalog_name: &str) { - assert!( - validation::is_authz_snapshot_stale(&err), - "expected a retryable stale-snapshot error, got: {err:#}" - ); - assert!( - err.to_string().contains(catalog_name), - "stale error should name the offending spec, got: {err:#}" - ); - } - /// With no capability filter the Snapshot is never consulted, so even a /// wholly unauthorized caller reading against a stale Snapshot gets the spec. /// This is the path controllers and other system callers take. diff --git a/crates/control-plane-api/src/publications/db_complete.rs b/crates/control-plane-api/src/publications/db_complete.rs index a00f19c0d09..e9f92e549c4 100644 --- a/crates/control-plane-api/src/publications/db_complete.rs +++ b/crates/control-plane-api/src/publications/db_complete.rs @@ -1,6 +1,6 @@ use crate::FlowType; -use super::{Capability, CatalogType, Id, TextJson as Json}; +use super::{CatalogType, Id, TextJson as Json}; use chrono::prelude::*; use serde::Serialize; @@ -449,25 +449,6 @@ pub async fn find_tenant_quotas( .await } -#[derive(Debug)] -pub struct ExpandedRow { - // Name of the specification. - pub catalog_name: String, - // Last build ID of the live spec. - pub last_build_id: Id, - // Last publication ID of the live spec. - pub last_pub_id: Id, - // Current live specification of this expansion. - // It won't be changed by this publication. - pub live_spec: Json>, - // ID of the expanded live specification. - pub live_spec_id: Id, - // Spec type of the live specification. - pub live_type: CatalogType, - // User's capability to the specification `catalog_name`. - pub user_capability: Option, -} - pub async fn delete_stale_flow( live_spec_id: Id, catalog_type: CatalogType, diff --git a/crates/control-plane-api/src/publications/specs.rs b/crates/control-plane-api/src/publications/specs.rs index ef4ac7a9a2d..3dd33121c44 100644 --- a/crates/control-plane-api/src/publications/specs.rs +++ b/crates/control-plane-api/src/publications/specs.rs @@ -1218,6 +1218,7 @@ mod test { #[cfg(test)] mod resolve_tests { use super::*; + use crate::test_support::{assert_stale_for, authoritative, published_at, stale}; // From `fixtures/authz_specs.sql`. const CAROL: uuid::Uuid = uuid::uuid!("33333333-3333-3333-3333-333333333333"); @@ -1268,39 +1269,6 @@ mod resolve_tests { })) } - /// Staleness compares the Snapshot's `taken` against the timestamp embedded - /// in a spec's `last_pub_id`, so read that back rather than recomputing it — - /// `flowid` is `macaddr8`, which silently widens short literals. - async fn published_at(pool: &sqlx::PgPool) -> tokens::DateTime { - sqlx::query_scalar!( - r#"select last_pub_id as "last_pub_id: models::Id" - from live_specs where catalog_name = $1"#, - COLLECTION, - ) - .fetch_one(pool) - .await - .expect("fixture collection should exist") - .timestamp() - } - - async fn snapshot_offset(pool: &sqlx::PgPool, offset: chrono::TimeDelta) -> crate::Snapshot { - let mut decrypted_hmac_keys = std::collections::HashMap::new(); - let data = crate::snapshot::try_fetch(pool, &mut decrypted_hmac_keys) - .await - .expect("failed to fetch snapshot"); - crate::Snapshot::new(published_at(pool).await + offset, data) - } - - /// Taken clear of the publication plus `TEMPORAL_SKEW`: denials are definitive. - async fn authoritative(pool: &sqlx::PgPool) -> crate::Snapshot { - snapshot_offset(pool, crate::Snapshot::TEMPORAL_SKEW * 4).await - } - - /// Taken before the publication it would judge: denials are retryable. - async fn stale(pool: &sqlx::PgPool) -> crate::Snapshot { - snapshot_offset(pool, -crate::Snapshot::TEMPORAL_SKEW * 4).await - } - /// Renders `live.errors` as `(scope, message)` pairs for snapshot assertions. fn error_pairs(live: &tables::LiveCatalog) -> Vec<(String, String)> { live.errors @@ -1309,17 +1277,6 @@ mod resolve_tests { .collect() } - fn assert_stale_for(err: anyhow::Error, catalog_name: &str) { - assert!( - validation::is_authz_snapshot_stale(&err), - "expected a retryable stale-snapshot error, got: {err:#}" - ); - assert!( - err.to_string().contains(catalog_name), - "stale error should name the offending spec, got: {err:#}" - ); - } - /// Branch 1: a user drafting an existing spec must admin it. Dan does not, /// but the denial is only definitive once the Snapshot outlives the spec. #[sqlx::test( 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 cacdd507887..2d5e8c798ba 100644 --- a/crates/control-plane-api/src/server/create_data_plane.rs +++ b/crates/control-plane-api/src/server/create_data_plane.rs @@ -1,7 +1,7 @@ -use crate::directives::storage_mappings::{fetch_storage_mappings, upsert_storage_mapping}; use crate::publications::{ DoNotRetry, DraftPublication, NoopInitialize, NoopWithCommit, PruneUnboundCollections, }; +use crate::storage_mappings::{fetch_storage_mappings, upsert_storage_mapping}; use anyhow::Context; use validator::Validate; @@ -64,19 +64,7 @@ pub async fn create_data_plane( }): super::Request, ) -> 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? - { - return Err(tonic::Status::permission_denied( - "authenticated user is not an admin of the 'ops/' tenant", - ) - .into()); - } + super::authorize_ops_admin(&env).await?; let (data_plane_fqdn, base_name, pulumi_stack) = match &private { None => ( @@ -232,7 +220,6 @@ pub async fn create_data_plane( .unwrap() .into(); - let snapshot = app.snapshot_watch.token(); let publication = DraftPublication { user_id: *user_id, logs_token: insert.logs_token, @@ -241,9 +228,7 @@ pub async fn create_data_plane( detail: Some(format!("publication for data-plane {base_name}")), // A one-shot handler invocation, with no queued row to anchor on. started_at: None, - snapshot: snapshot - .result() - .expect("authorization snapshot is not ready"), + snapshot: env.snapshot(), // We've already validated that the user can admin `ops/`, // so further authZ checks are unnecessary. verify_user_authz: false, @@ -332,3 +317,176 @@ impl Validate for Category { } } } + +/// The `ops/`-admin pre-check, evaluated against the request's pinned +/// Snapshot: a caller without `ops/` admin is rejected before any data-plane +/// state is touched — terminally (403) when the Snapshot postdates the +/// request, and with the platform-standard 307 retry when it doesn't. +#[cfg(test)] +mod test { + use crate::test_server; + + // From `fixtures/alice.sql`: admin of `aliceCo/` and nothing else. + const ALICE: uuid::Uuid = uuid::uuid!("11111111-1111-1111-1111-111111111111"); + + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "alice")) + )] + async fn test_create_data_plane_denied_for_non_ops_admin(pool: sqlx::PgPool) { + let _guard = test_server::init(); + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), false).await, + ) + .await; + let token = server.make_access_token(ALICE, Some("alice@example.com")); + + let response = server + .rest_client() + .post( + "/admin/create-data-plane", + &serde_json::json!({"name": "test-plane-c1", "category": "managed"}), + Some(&token), + ) + .send() + .await + .unwrap(); + + assert_eq!(reqwest::StatusCode::FORBIDDEN, response.status()); + } + + /// A denial evaluated against a Snapshot which predates the request is + /// provisional: the endpoint answers with the platform-standard 307 + /// `AuthZRetry` (Retry-After + `started`/`retryAfter` params) rather than + /// a terminal 403, and a retry against the refreshed (authoritative) + /// Snapshot then resolves the denial terminally. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "alice")) + )] + async fn test_create_data_plane_stale_snapshot_retries_then_denies(pool: sqlx::PgPool) { + let _guard = test_server::init(); + // gate=true serves an empty epoch-taken Snapshot first: every denial + // under it is provisional. The revoke cancelled by the first request + // refreshes the watch to the real (+1h) Snapshot. + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), true).await, + ) + .await; + let token = server.make_access_token(ALICE, Some("alice@example.com")); + + let client = flow_client_next::rest::Client { + base_url: server.base_url(), + http_client: reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap(), + }; + let body = serde_json::json!({"name": "test-plane-c1", "category": "managed"}); + + let response = client + .post("/admin/create-data-plane", &body, Some(&token)) + .send() + .await + .unwrap(); + assert_eq!( + reqwest::StatusCode::TEMPORARY_REDIRECT, + response.status(), + "a stale denial must be provisional" + ); + assert!( + response + .headers() + .contains_key(reqwest::header::RETRY_AFTER) + ); + let location = response + .headers() + .get(reqwest::header::LOCATION) + .and_then(|l| l.to_str().ok()) + .expect("redirect carries a Location"); + assert!( + location.contains("started=") && location.contains("retryAfter="), + "Location must carry retry bookkeeping: {location}" + ); + + // The cancelled revoke triggers a refresh to the authoritative + // Snapshot; denials then become terminal. Bound the wait, since the + // refresh races this retry loop. + for attempt in 0..50 { + let response = client + .post("/admin/create-data-plane", &body, Some(&token)) + .send() + .await + .unwrap(); + match response.status() { + reqwest::StatusCode::FORBIDDEN => return, + reqwest::StatusCode::TEMPORARY_REDIRECT => { + tokio::time::sleep(std::time::Duration::from_millis(20 * attempt)).await; + } + other => panic!("unexpected interim status {other}"), + } + } + panic!("denial never became terminal under the refreshed snapshot"); + } + + /// The seeded system user — these endpoints' routine caller — holds a + /// direct `('ops/', 'admin')` row in `user_grants` (seed.sql); this pins + /// that exactly that grant shape resolves through the Snapshot's grant + /// walk, and that it doesn't leak into unrelated tenants. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "alice")) + )] + async fn test_ops_admin_authorizes_via_snapshot(pool: sqlx::PgPool) { + let ops_admin = uuid::uuid!("99999999-9999-9999-9999-999999999999"); + sqlx::query("insert into auth.users (id, email) values ($1, 'ops-admin@example.com')") + .bind(ops_admin) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "insert into user_grants (user_id, object_role, capability) values ($1, 'ops/', 'admin')", + ) + .bind(ops_admin) + .execute(&pool) + .await + .unwrap(); + + let mut decrypted_hmac_keys = std::collections::HashMap::new(); + let data = crate::snapshot::try_fetch(&pool, &mut decrypted_hmac_keys) + .await + .expect("failed to fetch snapshot"); + let snapshot = crate::Snapshot::new(tokens::now(), data); + + let claims = models::authorizations::ControlClaims { + iat: 0, + exp: u64::MAX, + sub: ops_admin, + role: "authenticated".to_string(), + aud: "authenticated".to_string(), + email: Some("ops-admin@example.com".to_string()), + }; + assert!( + crate::evaluate_names_authorization( + &snapshot, + &claims, + models::Capability::Admin, + ["ops/"], + ) + .is_ok(), + "an ops/ admin user_grant must satisfy the snapshot walk" + ); + assert!( + crate::evaluate_names_authorization( + &snapshot, + &claims, + models::Capability::Admin, + ["aliceCo/"], + ) + .is_err(), + "ops/ admin must not leak into unrelated tenants" + ); + } +} diff --git a/crates/control-plane-api/src/server/mod.rs b/crates/control-plane-api/src/server/mod.rs index 406393688d2..19e36b9c045 100644 --- a/crates/control-plane-api/src/server/mod.rs +++ b/crates/control-plane-api/src/server/mod.rs @@ -134,6 +134,22 @@ where Ok((None, ())) } +/// Require that the requesting user holds `admin` over the `ops/` tenant, +/// evaluated against the request's pinned Snapshot — the same Snapshot the +/// rest of the operation then uses: one request, one authorization view. +/// A denial is terminal only once the Snapshot postdates the request's start; +/// otherwise `authorization_outcome` yields the standard retry response. +pub(crate) async fn authorize_ops_admin(env: &crate::Envelope) -> Result<(), crate::ApiError> { + let policy_result = crate::evaluate_names_authorization( + env.snapshot(), + env.claims()?, + models::Capability::Admin, + ["ops/"], + ); + let (_expiry, ()) = env.authorization_outcome(policy_result).await?; + Ok(()) +} + /// Looks up the user's authorization grants for each item in /// `prefixes_or_names`, and calls the provided `attach` function with each /// item and its capability. The `Some` results are returned in a vec. diff --git a/crates/control-plane-api/src/server/public/graphql/storage_mappings.rs b/crates/control-plane-api/src/server/public/graphql/storage_mappings.rs index 31ecd538405..6ebb7182f63 100644 --- a/crates/control-plane-api/src/server/public/graphql/storage_mappings.rs +++ b/crates/control-plane-api/src/server/public/graphql/storage_mappings.rs @@ -1,5 +1,5 @@ use super::filters; -use crate::directives::storage_mappings::{ +use crate::storage_mappings::{ collection_and_recovery_spec_from, insert_storage_mapping, strip_collection_data_suffix, update_storage_mapping, upsert_storage_mapping, }; diff --git a/crates/control-plane-api/src/server/snapshot.rs b/crates/control-plane-api/src/server/snapshot.rs index 1a2a52f05c4..9c9ddc69d74 100644 --- a/crates/control-plane-api/src/server/snapshot.rs +++ b/crates/control-plane-api/src/server/snapshot.rs @@ -258,6 +258,30 @@ impl Snapshot { ) } + /// Evaluate whether `user_id` may fetch the live spec `catalog_name` with + /// `capability`: the single spec-fetch policy shared by every live-spec + /// fetcher (`live_specs::get_live_specs`, `get_connected_live_specs`, + /// `evolutions::resolve_specs`). Staleness anchors on `started` — the + /// fetching operation's durable queued instant — when the caller has one, + /// and otherwise on the spec's own last publication, which bounds the + /// window in which grants could have been committed alongside the spec. + /// + /// `Ok(false)` is an authoritative denial: callers drop or suppress the + /// spec, the pre-existing behavior. A provisional denial surfaces as the + /// retryable `AuthorizationSnapshotStale` error instead. + pub fn spec_fetch_authorization( + &self, + user_id: uuid::Uuid, + catalog_name: &str, + capability: models::Capability, + started: Option, + last_pub_id: models::Id, + ) -> Result { + let anchor = started.unwrap_or_else(|| last_pub_id.timestamp()); + self.user_authorization(user_id, catalog_name, capability, Some(anchor)) + .ok_or_stale(catalog_name) + } + /// Evaluate whether `subject` (a catalog spec acting as a role) holds /// `capability` to `object` under this Snapshot's role grants, classified /// against `anchor` freshness (see `resolve_authorization`). 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 f7fc33ec5a0..253455fa6a7 100644 --- a/crates/control-plane-api/src/server/update_l2_reporting.rs +++ b/crates/control-plane-api/src/server/update_l2_reporting.rs @@ -33,19 +33,7 @@ pub async fn update_l2_reporting( }): super::Request, ) -> 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? - { - return Err(tonic::Status::permission_denied( - "authenticated user is not an admin of the 'ops/' tenant", - ) - .into()); - } + super::authorize_ops_admin(&env).await?; let template = include_str!("../../../../ops-catalog/reporting-L2-template.bundle.json"); let tables::DraftCatalog { collections, .. } = @@ -295,7 +283,6 @@ export class Derivation extends Types.IDerivation {"# }; let logs_token = uuid::Uuid::new_v4(); - let snapshot = app.snapshot_watch.token(); let publication = DraftPublication { user_id: *user_id, logs_token, @@ -304,9 +291,7 @@ export class Derivation extends Types.IDerivation {"# detail: Some(format!("publication for updating L2 reporting")), // A one-shot handler invocation, with no queued row to anchor on. started_at: None, - snapshot: snapshot - .result() - .expect("authorization snapshot is not ready"), + snapshot: env.snapshot(), default_data_plane_name: if default_data_plane.trim().is_empty() { None } else { @@ -352,6 +337,42 @@ export class Derivation extends Types.IDerivation {"# })) } +/// See `create_data_plane::test` — same pre-check, same contract: a caller +/// without `ops/` admin is rejected with 403 before any template work. +#[cfg(test)] +mod test { + use crate::test_server; + + const ALICE: uuid::Uuid = uuid::uuid!("11111111-1111-1111-1111-111111111111"); + + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "alice")) + )] + async fn test_update_l2_reporting_denied_for_non_ops_admin(pool: sqlx::PgPool) { + let _guard = test_server::init(); + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), false).await, + ) + .await; + let token = server.make_access_token(ALICE, Some("alice@example.com")); + + let response = server + .rest_client() + .post( + "/admin/update-l2-reporting", + &serde_json::json!({"defaultDataPlane": "", "dryRun": true}), + Some(&token), + ) + .send() + .await + .unwrap(); + + assert_eq!(reqwest::StatusCode::FORBIDDEN, response.status()); + } +} + // Copied from crates/derive-typescript/src/codegen/mod.rs fn camel_case(name: &str, mut upper: bool) -> String { let mut w = String::new(); diff --git a/crates/control-plane-api/src/directives/storage_mappings.rs b/crates/control-plane-api/src/storage_mappings.rs similarity index 96% rename from crates/control-plane-api/src/directives/storage_mappings.rs rename to crates/control-plane-api/src/storage_mappings.rs index 05ea027fdd2..f4887965cf7 100644 --- a/crates/control-plane-api/src/directives/storage_mappings.rs +++ b/crates/control-plane-api/src/storage_mappings.rs @@ -1,21 +1,12 @@ +//! Storage-mapping persistence helpers and the `collection-data/` prefix +//! conventions: the shared write path for `storage_mappings` rows, used by +//! the GraphQL storage-mapping resolvers and `create_data_plane`. Formerly +//! under `directives/`, until the GraphQL mutations superseded the +//! `storageMappings` applied-directive (2026-03-16) and the directive was +//! removed. + 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>, diff --git a/crates/control-plane-api/src/test_support.rs b/crates/control-plane-api/src/test_support.rs new file mode 100644 index 00000000000..6358df779ad --- /dev/null +++ b/crates/control-plane-api/src/test_support.rs @@ -0,0 +1,56 @@ +//! Shared helpers for authorization tests over the `authz_specs` fixture: +//! Snapshots of the database's real grants, stamped relative to the fixture's +//! publication instant so each test chooses whether a denial is authoritative +//! or provisional. + +/// The `authz_specs.sql` collection whose publication anchors staleness. +const ANCHOR_COLLECTION: &str = "carolCo/data/foo"; + +/// Staleness compares the Snapshot's `taken` against the timestamp embedded +/// in a spec's `last_pub_id`, so read that back rather than recomputing it — +/// `flowid` is `macaddr8`, which silently widens short literals. +pub(crate) async fn published_at(pool: &sqlx::PgPool) -> tokens::DateTime { + sqlx::query_scalar::<_, models::Id>( + "select last_pub_id from live_specs where catalog_name = $1", + ) + .bind(ANCHOR_COLLECTION) + .fetch_one(pool) + .await + .expect("fixture collection should exist") + .timestamp() +} + +/// A Snapshot holding the fixture's real grants, stamped `offset` away from +/// the instant the fixture's specs were published. +pub(crate) async fn snapshot_offset( + pool: &sqlx::PgPool, + offset: chrono::TimeDelta, +) -> crate::Snapshot { + let mut decrypted_hmac_keys = std::collections::HashMap::new(); + let data = crate::snapshot::try_fetch(pool, &mut decrypted_hmac_keys) + .await + .expect("failed to fetch snapshot"); + crate::Snapshot::new(published_at(pool).await + offset, data) +} + +/// Taken clear of the publication plus `TEMPORAL_SKEW`: denials are definitive. +pub(crate) async fn authoritative(pool: &sqlx::PgPool) -> crate::Snapshot { + snapshot_offset(pool, crate::Snapshot::TEMPORAL_SKEW * 4).await +} + +/// Taken before the publication it would judge: denials are retryable. +pub(crate) async fn stale(pool: &sqlx::PgPool) -> crate::Snapshot { + snapshot_offset(pool, -crate::Snapshot::TEMPORAL_SKEW * 4).await +} + +/// Asserts `err` is the retryable stale-snapshot error and names the spec. +pub(crate) fn assert_stale_for(err: anyhow::Error, catalog_name: &str) { + assert!( + validation::is_authz_snapshot_stale(&err), + "expected a retryable stale-snapshot error, got: {err:#}" + ); + assert!( + err.to_string().contains(catalog_name), + "stale error should name the offending spec, got: {err:#}" + ); +} diff --git a/local/systemd/flow-plane-link@.service b/local/systemd/flow-plane-link@.service index 8cc19f12821..802d70b6a3d 100644 --- a/local/systemd/flow-plane-link@.service +++ b/local/systemd/flow-plane-link@.service @@ -20,6 +20,11 @@ EnvironmentFile=%h/flow-local/env/plane-link-%i.env # during warm-up, bounded to 120s of 2s-spaced attempts. The agent binds its API # port only once it can serve (crates/agent/src/main.rs), so a connection being # accepted means ready; a non-transient 4xx still fails fast with its body. +# +# The -L flag follows the 307 AuthZRetry the /admin endpoints answer with when +# an authorization denial is evaluated against a not-yet-authoritative +# snapshot (307 preserves the POST and body; the server paces early retries +# itself). Without it curl treats the redirect as success and silently no-ops. ExecStart=/bin/bash -c ' \ if [ -n "${DEKAF_ADDRESS:-}" ] && [ -n "${DEKAF_REGISTRY_ADDRESS:-}" ]; then \ printf "{\\"name\\":\\"%%s\\",\\"category\\":{\\"manual\\":{\\"brokerAddress\\":\\"%%s\\",\\"reactorAddress\\":\\"%%s\\",\\"hmacKeys\\":[\\"%%s\\"],\\"dekafAddress\\":\\"%%s\\",\\"dekafRegistryAddress\\":\\"%%s\\"}}}" \ @@ -32,6 +37,8 @@ ExecStart=/bin/bash -c ' \ -X POST \ -H "content-type: application/json" \ -H "authorization: bearer ${SYSTEM_USER_TOKEN}" \ + -L \ + --max-redirs 10 \ --retry 60 \ --retry-connrefused \ --retry-delay 2 \ @@ -49,6 +56,8 @@ ExecStart=/bin/bash -c ' \ -X POST \ -H "content-type: application/json" \ -H "authorization: bearer ${SYSTEM_USER_TOKEN}" \ + -L \ + --max-redirs 10 \ --retry 60 \ --retry-connrefused \ --retry-delay 2 \ @@ -76,6 +85,8 @@ ExecStop=/bin/bash -c ' \ -X POST \ -H "content-type: application/json" \ -H "authorization: bearer ${SYSTEM_USER_TOKEN}" \ + -L \ + --max-redirs 10 \ --fail-with-body \ --data-binary @- \ ${AGENT_API}/admin/update-l2-reporting \