diff --git a/.sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json b/.sqlx/query-266d49e612499108de0c3742330a2e00b39906b01f43d835ec160751ed35c684.json similarity index 55% rename from .sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json rename to .sqlx/query-266d49e612499108de0c3742330a2e00b39906b01f43d835ec160751ed35c684.json index c1ba203be75..d401fe34493 100644 --- a/.sqlx/query-6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343.json +++ b/.sqlx/query-266d49e612499108de0c3742330a2e00b39906b01f43d835ec160751ed35c684.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n with collections(id) as (\n select ls.id\n from unnest($2::text[]) as names(catalog_name)\n join live_specs ls on ls.catalog_name = names.catalog_name\n ),\n exp(id) as (\n select lsf.source_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.target_id\n union\n select lsf.target_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.source_id\n )\n select\n ls.id as \"id: Id\",\n ls.last_pub_id as \"last_pub_id: Id\",\n ls.last_build_id as \"last_build_id: Id\",\n ls.data_plane_id as \"data_plane_id: Id\",\n ls.catalog_name,\n ls.spec_type as \"spec_type?: CatalogType\",\n ls.spec as \"spec: TextJson>\",\n ls.built_spec as \"built_spec: TextJson>\",\n ls.inferred_schema_md5,\n (\n select max(capability) from internal.user_roles($1) r\n where starts_with(ls.catalog_name, r.role_prefix)\n ) as \"user_capability: Capability\",\n coalesce(\n (select json_agg(row_to_json(role_grants))\n from role_grants\n where starts_with(ls.catalog_name, subject_role)),\n '[]'\n ) as \"spec_capabilities!: Json>\",\n ls.dependency_hash\n from exp\n join live_specs ls on ls.id = exp.id\n where ls.spec is not null and not ls.catalog_name = any($3);\n ", + "query": "\n with collections(id) as (\n select ls.id\n from unnest($1::text[]) as names(catalog_name)\n join live_specs ls on ls.catalog_name = names.catalog_name\n ),\n exp(id) as (\n select lsf.source_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.target_id\n union\n select lsf.target_id as id\n from collections c\n join live_spec_flows lsf on c.id = lsf.source_id\n )\n select\n ls.id as \"id: Id\",\n ls.last_pub_id as \"last_pub_id: Id\",\n ls.last_build_id as \"last_build_id: Id\",\n ls.data_plane_id as \"data_plane_id: Id\",\n ls.catalog_name,\n ls.spec_type as \"spec_type?: CatalogType\",\n ls.spec as \"spec: TextJson>\",\n ls.built_spec as \"built_spec: TextJson>\",\n ls.inferred_schema_md5,\n ls.dependency_hash\n from exp\n join live_specs ls on ls.id = exp.id\n where ls.spec is not null and not ls.catalog_name = any($2);\n ", "describe": { "columns": [ { @@ -62,62 +62,12 @@ }, { "ordinal": 9, - "name": "user_capability: Capability", - "type_info": { - "Custom": { - "name": "grant_capability", - "kind": { - "Enum": [ - "none", - "x_01", - "x_02", - "x_03", - "x_04", - "x_05", - "x_06", - "x_07", - "x_08", - "x_09", - "read", - "x_11", - "x_12", - "x_13", - "x_14", - "x_15", - "x_16", - "x_17", - "x_18", - "x_19", - "write", - "x_21", - "x_22", - "x_23", - "x_24", - "x_25", - "x_26", - "x_27", - "x_28", - "x_29", - "admin" - ] - } - } - } - }, - { - "ordinal": 10, - "name": "spec_capabilities!: Json>", - "type_info": "Json" - }, - { - "ordinal": 11, "name": "dependency_hash", "type_info": "Text" } ], "parameters": { "Left": [ - "Uuid", "TextArray", "TextArray" ] @@ -132,10 +82,8 @@ true, true, true, - null, - null, true ] }, - "hash": "6bc21fd940535409a6b59d230fed417ff92af5cb2f6c11a35944288b14fde343" + "hash": "266d49e612499108de0c3742330a2e00b39906b01f43d835ec160751ed35c684" } diff --git a/.sqlx/query-34dcc2c093232d6e00d1192a46fb2aedf85caf5c5625373ddd4311601dd1fa9c.json b/.sqlx/query-34dcc2c093232d6e00d1192a46fb2aedf85caf5c5625373ddd4311601dd1fa9c.json new file mode 100644 index 00000000000..99b7fa35208 --- /dev/null +++ b/.sqlx/query-34dcc2c093232d6e00d1192a46fb2aedf85caf5c5625373ddd4311601dd1fa9c.json @@ -0,0 +1,88 @@ +{ + "db_name": "PostgreSQL", + "query": "\n select\n coalesce(ls.id, '00:00:00:00:00:00:00:00'::flowid) as \"id!: Id\",\n coalesce(ls.last_pub_id, '00:00:00:00:00:00:00:00'::flowid) as \"last_pub_id!: Id\",\n coalesce(ls.last_build_id, '00:00:00:00:00:00:00:00'::flowid) as \"last_build_id!: Id\",\n coalesce(ls.data_plane_id, '00:00:00:00:00:00:00:00'::flowid) as \"data_plane_id!: Id\",\n names as \"catalog_name!: String\",\n ls.spec_type as \"spec_type?: CatalogType\",\n ls.spec as \"spec: TextJson>\",\n ls.built_spec as \"built_spec: TextJson>\",\n ls.inferred_schema_md5,\n ls.dependency_hash\n from unnest($1::text[]) names\n left outer join live_specs ls on ls.catalog_name = names\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "last_pub_id!: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 2, + "name": "last_build_id!: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 3, + "name": "data_plane_id!: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 4, + "name": "catalog_name!: String", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "spec_type?: CatalogType", + "type_info": { + "Custom": { + "name": "catalog_spec_type", + "kind": { + "Enum": [ + "capture", + "collection", + "materialization", + "test" + ] + } + } + } + }, + { + "ordinal": 6, + "name": "spec: TextJson>", + "type_info": "Json" + }, + { + "ordinal": 7, + "name": "built_spec: TextJson>", + "type_info": "Json" + }, + { + "ordinal": 8, + "name": "inferred_schema_md5", + "type_info": "Text" + }, + { + "ordinal": 9, + "name": "dependency_hash", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "TextArray" + ] + }, + "nullable": [ + null, + null, + null, + null, + null, + true, + true, + true, + true, + true + ] + }, + "hash": "34dcc2c093232d6e00d1192a46fb2aedf85caf5c5625373ddd4311601dd1fa9c" +} diff --git a/.sqlx/query-6336a73b8d0dedacb47563fc753ca867a6e5a285c2862d552910767381170ab0.json b/.sqlx/query-6336a73b8d0dedacb47563fc753ca867a6e5a285c2862d552910767381170ab0.json deleted file mode 100644 index 84aea275149..00000000000 --- a/.sqlx/query-6336a73b8d0dedacb47563fc753ca867a6e5a285c2862d552910767381170ab0.json +++ /dev/null @@ -1,142 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n with user_roles as materialized (\n select role_prefix, capability from internal.user_roles($1)\n )\n select\n coalesce(ls.id, '00:00:00:00:00:00:00:00'::flowid) as \"id!: Id\",\n coalesce(ls.last_pub_id, '00:00:00:00:00:00:00:00'::flowid) as \"last_pub_id!: Id\",\n coalesce(ls.last_build_id, '00:00:00:00:00:00:00:00'::flowid) as \"last_build_id!: Id\",\n coalesce(ls.data_plane_id, '00:00:00:00:00:00:00:00'::flowid) as \"data_plane_id!: Id\",\n names as \"catalog_name!: String\",\n ls.spec_type as \"spec_type?: CatalogType\",\n ls.spec as \"spec: TextJson>\",\n ls.built_spec as \"built_spec: TextJson>\",\n ls.inferred_schema_md5,\n case when $3 then (\n select max(capability) from user_roles\n where starts_with(names, user_roles.role_prefix)\n ) else\n null\n end as \"user_capability: Capability\",\n case when $4 then coalesce(\n (select json_agg(row_to_json(role_grants))\n from role_grants\n where starts_with(names, subject_role)),\n '[]'\n ) else\n '[]'\n end as \"spec_capabilities!: Json>\",\n ls.dependency_hash\n from unnest($2::text[]) names\n left outer join live_specs ls on ls.catalog_name = names\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id!: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 1, - "name": "last_pub_id!: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 2, - "name": "last_build_id!: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 3, - "name": "data_plane_id!: Id", - "type_info": "Macaddr8" - }, - { - "ordinal": 4, - "name": "catalog_name!: String", - "type_info": "Text" - }, - { - "ordinal": 5, - "name": "spec_type?: CatalogType", - "type_info": { - "Custom": { - "name": "catalog_spec_type", - "kind": { - "Enum": [ - "capture", - "collection", - "materialization", - "test" - ] - } - } - } - }, - { - "ordinal": 6, - "name": "spec: TextJson>", - "type_info": "Json" - }, - { - "ordinal": 7, - "name": "built_spec: TextJson>", - "type_info": "Json" - }, - { - "ordinal": 8, - "name": "inferred_schema_md5", - "type_info": "Text" - }, - { - "ordinal": 9, - "name": "user_capability: Capability", - "type_info": { - "Custom": { - "name": "grant_capability", - "kind": { - "Enum": [ - "none", - "x_01", - "x_02", - "x_03", - "x_04", - "x_05", - "x_06", - "x_07", - "x_08", - "x_09", - "read", - "x_11", - "x_12", - "x_13", - "x_14", - "x_15", - "x_16", - "x_17", - "x_18", - "x_19", - "write", - "x_21", - "x_22", - "x_23", - "x_24", - "x_25", - "x_26", - "x_27", - "x_28", - "x_29", - "admin" - ] - } - } - } - }, - { - "ordinal": 10, - "name": "spec_capabilities!: Json>", - "type_info": "Json" - }, - { - "ordinal": 11, - "name": "dependency_hash", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Uuid", - "TextArray", - "Bool", - "Bool" - ] - }, - "nullable": [ - null, - null, - null, - null, - null, - true, - true, - true, - true, - null, - null, - true - ] - }, - "hash": "6336a73b8d0dedacb47563fc753ca867a6e5a285c2862d552910767381170ab0" -} diff --git a/.sqlx/query-87101040b0795647b0b9f144f24dc199f724caf45a9d9755a9fdc9fba9dc05e2.json b/.sqlx/query-e12384cbb271c0adffbeea930e0435851d0b21a93f5705f592796a5abe1542b5.json similarity index 63% rename from .sqlx/query-87101040b0795647b0b9f144f24dc199f724caf45a9d9755a9fdc9fba9dc05e2.json rename to .sqlx/query-e12384cbb271c0adffbeea930e0435851d0b21a93f5705f592796a5abe1542b5.json index c29fbf78e4b..70c86e1ebbf 100644 --- a/.sqlx/query-87101040b0795647b0b9f144f24dc199f724caf45a9d9755a9fdc9fba9dc05e2.json +++ b/.sqlx/query-e12384cbb271c0adffbeea930e0435851d0b21a93f5705f592796a5abe1542b5.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n WITH\n data_plane_ids AS (\n SELECT id\n FROM UNNEST($1::flowid[]) AS t(id)\n ),\n data_plane_names AS (\n SELECT name\n FROM UNNEST($2::text[]) AS t(name)\n -- User must be read-authorized to data-plane.\n WHERE EXISTS (\n SELECT 1\n FROM internal.user_roles($3, 'read') AS r\n WHERE starts_with(t.name, r.role_prefix)\n )\n )\n SELECT\n d.id AS \"control_id: Id\",\n d.data_plane_name,\n d.closed,\n d.hmac_keys,\n d.encrypted_hmac_keys AS \"encrypted_hmac_keys: models::RawValue\",\n d.data_plane_fqdn,\n d.broker_address,\n d.reactor_address,\n d.dekaf_address,\n d.dekaf_registry_address,\n d.ops_logs_name AS \"ops_logs_name: models::Collection\",\n d.ops_stats_name AS \"ops_stats_name: models::Collection\"\n FROM data_planes d\n WHERE\n d.id IN (select id from data_plane_ids) OR\n d.data_plane_name in (select name from data_plane_names)\n ", + "query": "\n WITH\n data_plane_ids AS (\n SELECT id\n FROM UNNEST($1::flowid[]) AS t(id)\n ),\n data_plane_names AS (\n -- Names have already passed the caller's user-authorization policy,\n -- so no in-SQL authorization check is needed here.\n SELECT name\n FROM UNNEST($2::text[]) AS t(name)\n )\n SELECT\n d.id AS \"control_id: Id\",\n d.data_plane_name,\n d.closed,\n d.hmac_keys,\n d.encrypted_hmac_keys AS \"encrypted_hmac_keys: models::RawValue\",\n d.data_plane_fqdn,\n d.broker_address,\n d.reactor_address,\n d.dekaf_address,\n d.dekaf_registry_address,\n d.ops_logs_name AS \"ops_logs_name: models::Collection\",\n d.ops_stats_name AS \"ops_stats_name: models::Collection\"\n FROM data_planes d\n WHERE\n d.id IN (select id from data_plane_ids) OR\n d.data_plane_name in (select name from data_plane_names)\n ", "describe": { "columns": [ { @@ -81,8 +81,7 @@ } } }, - "TextArray", - "Uuid" + "TextArray" ] }, "nullable": [ @@ -100,5 +99,5 @@ false ] }, - "hash": "87101040b0795647b0b9f144f24dc199f724caf45a9d9755a9fdc9fba9dc05e2" + "hash": "e12384cbb271c0adffbeea930e0435851d0b21a93f5705f592796a5abe1542b5" } diff --git a/crates/agent/src/controlplane.rs b/crates/agent/src/controlplane.rs index fb14799a0b9..ce2b1bda85b 100644 --- a/crates/agent/src/controlplane.rs +++ b/crates/agent/src/controlplane.rs @@ -577,12 +577,16 @@ impl ControlPlane for PGControlPlane } async fn get_live_specs(&self, names: BTreeSet) -> anyhow::Result { + let snapshot = self.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let names = names.into_iter().collect::>(); let mut live = live_specs::get_live_specs( self.system_user_id, &names, None, // don't filter based on user capability &self.pool, + &snapshot, + None, ) .await?; @@ -655,6 +659,9 @@ impl ControlPlane for PGControlPlane logs_token, data_plane: data_plane.clone(), created_at, + snapshot, + // `filter_user_authz` is false, so no staleness anchor is consulted. + started_at: None, }; discovers_handler.discover(pool, req).await } @@ -666,6 +673,7 @@ impl ControlPlane for PGControlPlane draft: tables::DraftCatalog, default_data_plane: Option, ) -> anyhow::Result { + let snapshot = self.snapshot_watch.token(); let publication = DraftPublication { user_id: self.system_user_id, logs_token, @@ -673,7 +681,14 @@ impl ControlPlane for PGControlPlane detail, dry_run: false, default_data_plane_name: default_data_plane, - // skip authz checks for controller-initiated publications + // Controllers construct a fresh publication per poll, so they have + // no instant that stays fixed across attempts to anchor staleness + // on; they carry their own retry/backoff instead. + started_at: None, + snapshot: snapshot + .result() + .expect("authorization snapshot is not ready"), + // Skip user-to-catalog checks; spec-to-spec `RoleGrant` checks remain mandatory. verify_user_authz: false, initialize: NoopInitialize, finalize: PruneUnboundCollections, diff --git a/crates/agent/src/discovers.rs b/crates/agent/src/discovers.rs index a0bfdbc3583..23d7e8e423b 100644 --- a/crates/agent/src/discovers.rs +++ b/crates/agent/src/discovers.rs @@ -1,6 +1,6 @@ use anyhow::Context; use control_plane_api::{ - connector_tags, + Snapshot, connector_tags, discovers::{Discover, DiscoverHandler, Row, fetch_discover}, draft, live_specs, proxy_connectors::DiscoverConnectors, @@ -100,6 +100,9 @@ fn precheck_failed(status: JobStatus) -> (JobStatus, ProcessResult) { pub struct DiscoverExecutor { pub handler: DiscoverHandler, + /// Authorization Snapshot watch. Each poll pins one Snapshot from this + /// watch, which serves every authorization decision of the discover. + pub snapshot_watch: std::sync::Arc>, } impl automations::Executor for DiscoverExecutor { @@ -127,7 +130,13 @@ impl automations::Executor for DiscoverExecutor { let draft_id = row.draft_id; assert_eq!(row.id, task_id); let time_queued = chrono::Utc::now().signed_duration_since(row.updated_at); - let (status, result) = self.process(row, pool).await?; + + // Pin one Snapshot for this poll: every authorization decision of the + // discover observes the same view. + let snapshot = self.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); + + let (status, result) = self.process(row, pool, snapshot).await?; tracing::info!(id=%task_id, %time_queued, ?status, "finished"); inbox.clear(); Ok(DiscoverOutcome { @@ -145,6 +154,7 @@ impl DiscoverExecutor { &self, row: Row, pool: &sqlx::PgPool, + snapshot: &Snapshot, ) -> anyhow::Result<(JobStatus, ProcessResult)> { tracing::info!( %row.capture_name, @@ -216,6 +226,12 @@ impl DiscoverExecutor { image_composed, data_plane, pool, + &snapshot, + // `None` anchors authorization staleness to each spec's own + // `last_pub_id`, preserving the pre-Snapshot semantics. A + // follow-up anchors this to the queued discover row and defers + // on staleness instead. + None, ) .await; @@ -263,7 +279,7 @@ impl DiscoverExecutor { /// row, even if it differs from the endpoint on the drafted or live spec. All /// other specs in the given draft will be loaded as they are and used as the /// base for the merge after the discover completes. -async fn prepare_discover( +async fn prepare_discover<'a>( user_id: uuid::Uuid, draft_id: Id, capture_name: models::Capture, @@ -272,8 +288,10 @@ async fn prepare_discover( logs_token: uuid::Uuid, image_composed: String, data_plane: tables::DataPlane, - pool: &sqlx::PgPool, -) -> anyhow::Result { + pool: &'a sqlx::PgPool, + snapshot: &'a Snapshot, + started_at: Option, +) -> anyhow::Result> { let mut draft = draft::load_draft(draft_id, pool) .await .context("loading draft")?; @@ -287,11 +305,18 @@ async fn prepare_discover( // embedded in its control-plane Id — is carried on the Discover request, // so that re-discovers resolve connector feature-flag defaults as the // running task does. It's empty for a task which doesn't exist yet. - // Filter to only specs that the user can read. If they can't admin, then + // Filter to only specs that the user can view. If they can't edit, then // wait until they try to publish to surface that error. let name = &[capture_name.to_string()]; - let live = - live_specs::get_live_specs(user_id, name, Some(models::Capability::Read), pool).await?; + let live = live_specs::get_live_specs( + user_id, + name, + Some(models::authz::Capability::CatalogRead.into()), + pool, + &snapshot, + started_at, + ) + .await?; let live_capture = live.captures.into_iter().next(); let created_at = live_capture .as_ref() @@ -362,6 +387,8 @@ async fn prepare_discover( reset_on_key_change, logs_token, created_at, + snapshot, + started_at, }) } @@ -442,7 +469,10 @@ mod test { dekaf_address: None, dekaf_registry_address: None, }; - + harness.refresh_snapshot().await; + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); + let started_at = tokens::now(); let result = super::prepare_discover( user_id, draft_id, @@ -453,6 +483,8 @@ mod test { image_composed.clone(), data_plane.clone(), &harness.pool, + &snapshot, + Some(started_at), ) .await .unwrap(); diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index 47a9ca2f757..c3f90ed5482 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -177,6 +177,11 @@ pub struct TestHarness { pub test_name: String, pub pool: sqlx::PgPool, pub publisher: Publisher, + /// Live authorization Snapshot watch. See the Snapshot testing model + /// documented above `fetch_snapshot`. + pub snapshot_watch: Arc>, + /// Manual write handle backing `snapshot_watch`; same reference. + set_snapshot: Arc, #[allow(dead_code)] // only here so we don't drop it until the harness is dropped pub builds_root: tempfile::TempDir, pub discover_handler: DiscoverHandler, @@ -241,6 +246,16 @@ impl HarnessBuilder { eprintln!("end of PUB-LOG"); }); + // Back the authorization Snapshot with a manually-driven watch (see + // the Snapshot testing model documented above `fetch_snapshot`). + let (snapshot_pending, snapshot_replace) = tokens::manual::(); + let set_snapshot: Arc = + Arc::new(move |snapshot| { + _ = snapshot_replace(Ok(snapshot)); + }); + set_snapshot(TestHarness::fetch_snapshot(&pool).await); + let snapshot_watch = snapshot_pending.ready_owned().await; + let mock_connectors = connectors::MockDiscoverConnectors::default(); let discover_handler = DiscoverHandler::new(mock_connectors.clone()); @@ -256,16 +271,13 @@ impl HarnessBuilder { ) .with_skip_all_tests(); - let snapshot_source = control_plane_api::snapshot::PgSnapshotSource::new(pool.clone()); - let snapshot_watch = tokens::watch(snapshot_source).ready_owned().await; - let control_plane = TestControlPlane::new(PGControlPlane::new( pool.clone(), system_user_id, publisher.clone(), discover_handler.clone(), logs_tx.clone(), - snapshot_watch, + snapshot_watch.clone(), 1.0, // auto_discover_probability publication_cooldown, crate::controllers::ControllerConfig::default(), @@ -280,6 +292,8 @@ impl HarnessBuilder { test_name, pool, publisher, + snapshot_watch, + set_snapshot, builds_root, discover_handler, control_plane, @@ -293,6 +307,9 @@ impl HarnessBuilder { }; harness.truncate_tables().await; harness.setup_test_connectors().await; + // The Snapshot was taken before `truncate_tables` cleared grants; re-fetch + // so authorization sees the truncated baseline rather than stale grants. + harness.refresh_snapshot().await; harness } @@ -484,6 +501,14 @@ impl TestHarness { del_tenants as ( delete from tenants ), + -- Storage mappings must be cleared too: `provision_tenant` (and the + -- beta-onboard directive) insert a tenant's mapping with `on conflict + -- do nothing`, so a mapping left over from an earlier run — including + -- one whose `data_planes` captured a developer's live local stack — + -- would silently survive and be read by the next test. + del_storage_mappings as ( + delete from storage_mappings + ), del_user_grants as ( -- preserve the system user's role grants delete from user_grants where user_id != $1 @@ -523,6 +548,19 @@ impl TestHarness { ), del_daily_stats as ( delete from catalog_stats_daily + ), + -- Clear data-planes too, so every test starts from a deterministic + -- baseline regardless of any data-planes a developer's live local + -- stack has registered in this shared database (e.g. a running + -- `mise run local:stack` registers `ops/dp/public/-cluster`). + -- `setup_test_connectors` re-inserts the single `ops/dp/public/test` + -- plane the tests expect. `data_plane_private_links` is deleted first + -- to satisfy its foreign key onto `data_planes`. + del_data_plane_private_links as ( + delete from internal.data_plane_private_links + ), + del_data_planes as ( + delete from data_planes ) delete from catalog_stats_monthly;"#, system_user_id @@ -547,6 +585,59 @@ impl TestHarness { &mut self.control_plane } + // The harness's Snapshot testing model: + // + // Production refreshes the authorization Snapshot through + // `PgSnapshotSource`'s timer-gated polling loop. The harness backs the + // same watch with a manual writer (`set_snapshot`) instead: refreshes are + // explicit, nothing refreshes on a timer, and `MIN_REFRESH_INTERVAL` + // never gates a test. Grant-mutating helpers refresh the watch after + // writing, so authorization observes what a test just set up. + + // Current Postgres state, stamped `taken = now()`. + async fn fetch_snapshot(pool: &sqlx::PgPool) -> control_plane_api::Snapshot { + Self::fetch_snapshot_at(pool, tokens::now()).await + } + + // Current Postgres state with a caller-chosen `taken` — the same query + // `PgSnapshotSource` runs, minus its cool-off. + async fn fetch_snapshot_at( + pool: &sqlx::PgPool, + taken: tokens::DateTime, + ) -> control_plane_api::Snapshot { + let mut decrypted_hmac_keys = std::collections::HashMap::new(); + let data = control_plane_api::snapshot::try_fetch(pool, &mut decrypted_hmac_keys) + .await + .expect("failed to fetch authorization snapshot"); + control_plane_api::Snapshot::new(taken, data) + } + + /// Re-fetches the Snapshot at `taken = now()`, making grant changes + /// written directly to Postgres visible. + pub async fn refresh_snapshot(&self) { + self.refresh_snapshot_at(tokens::now()).await + } + + /// Refreshes with an exact `taken`. + pub async fn refresh_snapshot_at(&self, taken: tokens::DateTime) { + let snapshot = Self::fetch_snapshot_at(&self.pool, taken).await; + (self.set_snapshot)(snapshot); + } + + /// Refreshes with `taken` pushed far enough forward to be authoritative + /// for everything written up to now: any denial it produces is definitive + /// rather than retryable. + pub async fn refresh_snapshot_authoritative(&self) { + self.refresh_snapshot_at(tokens::now() + Self::snapshot_settle()) + .await + } + + // Margin pushing `taken` clear of `TEMPORAL_SKEW` in either direction. + // Any multiple > 1 works; 4 leaves obvious headroom. + fn snapshot_settle() -> chrono::TimeDelta { + control_plane_api::Snapshot::TEMPORAL_SKEW * 4 + } + /// Setup a new tenant with the given name, and return the id of the user /// who has `admin` capabilities to it. Performs essentially the same setup /// as the beta onboarding directive, so the user_grants, role_grants, @@ -559,10 +650,13 @@ impl TestHarness { "full_name": format!("Full ({tenant}) Name"), }); - control_plane_api::directives::beta_onboard::provision_test_tenant( + let user_id = control_plane_api::directives::beta_onboard::provision_test_tenant( &self.pool, tenant, &email, meta, ) - .await + .await; + // Grants were just written; re-sync the authorization Snapshot. + self.refresh_snapshot().await; + user_id } pub async fn add_role_grant(&mut self, subject: &str, object: &str, capability: Capability) { @@ -578,6 +672,8 @@ impl TestHarness { .execute(&self.pool) .await .unwrap(); + // Re-sync the authorization Snapshot with the new grant. + self.refresh_snapshot().await; } pub async fn add_user_grant(&mut self, user_id: Uuid, role: &str, capability: Capability) { @@ -592,23 +688,19 @@ impl TestHarness { .await .unwrap(); txn.commit().await.unwrap(); + // Re-sync the authorization Snapshot with the new grant. + self.refresh_snapshot().await; } pub async fn assert_specs_touched_since(&mut self, prev_specs: &tables::LiveCatalog) { - let user_id = self.control_plane().inner.system_user_id; let owned_names: Vec = prev_specs .all_spec_names() .map(|n| (*n).to_owned()) .collect(); - let specs = control_plane_api::live_specs::fetch_live_specs( - user_id, - &owned_names, - false, /* don't fetch user capabilities */ - false, /* don't fetch spec capabilities */ - &self.pool, - ) - .await - .expect("failed to query live specs"); + + let specs = control_plane_api::live_specs::fetch_live_specs(&owned_names, &self.pool) + .await + .expect("failed to query live specs"); assert_eq!( prev_specs.spec_count(), specs.len(), @@ -1130,12 +1222,14 @@ impl TestHarness { task_types::PUBLICATIONS => Server::new().register(PublicationsExecutor { publisher: self.publisher.clone(), pg_pool: self.pool.clone(), + snapshot_watch: self.snapshot_watch.clone(), runtime_v2_new_captures: self.runtime_v2_new_captures, runtime_v2_new_materializations: self.runtime_v2_new_materializations, runtime_v2_new_derivations: self.runtime_v2_new_derivations, }), task_types::DISCOVERS => Server::new().register(DiscoverExecutor { handler: self.discover_handler.clone(), + snapshot_watch: self.snapshot_watch.clone(), }), task_types::APPLIED_DIRECTIVES => Server::new().register(self.directive_exec.clone()), task_types::TENANT_ALERT_EVALS => Server::new().register( @@ -1326,6 +1420,12 @@ impl TestHarness { .expect("failed to create publication"); txn.commit().await.expect("failed to commit transaction"); + // The publication's pinned Snapshot must be authoritative for specs + // committed earlier in this test, or an authorization denial reads as + // provisional. Compressed test time never advances past the skew on + // its own, so model production's elapsed wait explicitly. + self.refresh_snapshot_authoritative().await; + let task_id = self .run_automation_task(automations::task_types::PUBLICATIONS) .await @@ -2139,6 +2239,7 @@ impl ControlPlane for TestControlPlane { let mocks = self.mocks.lock().unwrap(); mocks.build_failures.clone() }; + let snapshot = self.inner.snapshot_watch.token(); let publication = DraftPublication { user_id: self.inner.system_user_id, detail, @@ -2146,6 +2247,11 @@ impl ControlPlane for TestControlPlane { logs_token, dry_run: false, default_data_plane_name: data_plane_name, + // Mirrors the production controller path, which has no queued row. + started_at: None, + snapshot: snapshot + .result() + .expect("authorization snapshot is not ready"), verify_user_authz: false, initialize: NoopInitialize, finalize, diff --git a/crates/agent/src/integration_tests/locking_retries.rs b/crates/agent/src/integration_tests/locking_retries.rs index af66b7b4749..b28d18b5553 100644 --- a/crates/agent/src/integration_tests/locking_retries.rs +++ b/crates/agent/src/integration_tests/locking_retries.rs @@ -23,7 +23,8 @@ async fn test_publication_concurrent_commits() { "beavers/dams": minimal_capture(None, &["beavers/dens"]), } })); - + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); // Try to reproduce a scenario where multiple different publications all try to commit // concurrently. We'll expect exactly one of them to succeed, and the others to fail. for test_iteration in 0..5 { @@ -41,6 +42,8 @@ async fn test_publication_concurrent_commits() { None, true, 0, + &snapshot, + None, ) .await .unwrap(); @@ -55,6 +58,8 @@ async fn test_publication_concurrent_commits() { None, true, 0, + &snapshot, + None, ) .await .unwrap(); @@ -68,6 +73,8 @@ async fn test_publication_concurrent_commits() { None, true, 0, + &snapshot, + None, ) .await .unwrap(); @@ -110,6 +117,8 @@ async fn test_publication_optimistic_locking_failures() { "mice/also-new": minimal_capture(Some(Id::new([8, 7, 6, 5, 4, 3, 2, 1])), &["mice/does-not-exist"]), } })); + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let naughty_pub_id = Id::new([8; 8]); // If a user explicitly sets `expectPubId` in the model, then a mismatch gets returned as a // build error, before we even try to commit. @@ -124,6 +133,8 @@ async fn test_publication_optimistic_locking_failures() { None, true, 0, + &snapshot, + None, ) .await .expect("build failed"); @@ -157,7 +168,8 @@ async fn test_publication_optimistic_locking_failures() { "mice/capture": minimal_capture(None, &["mice/cheese", "mice/seeds"]), } }); - + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let will_fail_pub = Id::new([9; 8]); let will_fail_build = harness .publisher @@ -170,10 +182,13 @@ async fn test_publication_optimistic_locking_failures() { None, true, 0, + &snapshot, + None, ) .await .expect("build a failed"); - + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let will_commit_pub = Id::new([10; 8]); let will_commit_build = harness .publisher @@ -186,6 +201,8 @@ async fn test_publication_optimistic_locking_failures() { None, true, 0, + snapshot, + None, ) .await .expect("build b failed"); @@ -228,6 +245,8 @@ async fn test_publication_optimistic_locking_failures() { "mice/capture": minimal_capture(None, &["mice/cheese", "mice/seeds"]), } })); + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let will_fail_build = harness .publisher .build( @@ -239,6 +258,8 @@ async fn test_publication_optimistic_locking_failures() { None, true, 0, + snapshot, + None, ) .await .expect("cheese build failed"); @@ -253,6 +274,8 @@ async fn test_publication_optimistic_locking_failures() { "mice/capture": minimal_capture(None, &["mice/cheese", "mice/seeds"]), } })); + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let will_commit_build = harness .publisher .build( @@ -264,6 +287,8 @@ async fn test_publication_optimistic_locking_failures() { None, true, 0, + snapshot, + None, ) .await .expect("seeds build failed"); @@ -402,6 +427,8 @@ async fn test_injected_ops_collections_are_not_locked() { "owls/capture": minimal_capture(None, &["owls/hoots"]), } })); + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let build = harness .publisher .build( @@ -413,6 +440,8 @@ async fn test_injected_ops_collections_are_not_locked() { None, true, 0, + snapshot, + None, ) .await .expect("owls build failed"); @@ -498,6 +527,8 @@ async fn test_injected_ops_collections_are_not_locked() { } } })); + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let reader_build = harness .publisher .build( @@ -509,6 +540,8 @@ async fn test_injected_ops_collections_are_not_locked() { None, true, 0, + snapshot, + None, ) .await .expect("reader build failed"); diff --git a/crates/agent/src/integration_tests/source_captures.rs b/crates/agent/src/integration_tests/source_captures.rs index fdd47013444..0e9778eb7ce 100644 --- a/crates/agent/src/integration_tests/source_captures.rs +++ b/crates/agent/src/integration_tests/source_captures.rs @@ -420,10 +420,24 @@ async fn test_source_capture_no_annotations() { } } })); + + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let pub_id = Id::new([0, 0, 0, 0, 0, 0, 0, 9]); let built = harness .publisher - .build(user_id, pub_id, None, draft, Uuid::new_v4(), None, false, 0) + .build( + user_id, + pub_id, + None, + draft, + Uuid::new_v4(), + None, + false, + 0, + snapshot, + None, + ) .await .expect("build failed"); assert!(built.has_errors()); diff --git a/crates/agent/src/integration_tests/unknown_connectors.rs b/crates/agent/src/integration_tests/unknown_connectors.rs index 0de6930a24b..8c167f42e21 100644 --- a/crates/agent/src/integration_tests/unknown_connectors.rs +++ b/crates/agent/src/integration_tests/unknown_connectors.rs @@ -41,10 +41,23 @@ async fn test_forbidden_connector() { } } })); + let snapshot = harness.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); let pub_id = Id::new([0, 0, 0, 0, 0, 0, 0, 9]); let built = harness .publisher - .build(user_id, pub_id, None, draft, Uuid::new_v4(), None, true, 0) + .build( + user_id, + pub_id, + None, + draft, + Uuid::new_v4(), + None, + true, + 0, + snapshot, + None, + ) .await .expect("build failed"); assert!(built.has_errors()); diff --git a/crates/agent/src/main.rs b/crates/agent/src/main.rs index dab23d93311..668d6f2eb8b 100644 --- a/crates/agent/src/main.rs +++ b/crates/agent/src/main.rs @@ -314,6 +314,20 @@ async fn async_main(args: Args) -> Result<(), anyhow::Error> { .context("failed to create builds-root directory")?; } + // Create the snapshot source and start the refresh loop. + // Snapshot fetches retry internally forever, so a persistent failure (a + // broken query, sops / KMS breakage) would otherwise hang here with the + // port unbound and nothing logged at error level. Bound the wait so that + // startup fails visibly, and fits within Cloud Run's 240s startup probe + // window even after the database retry budget above. + let snapshot_source = control_plane_api::snapshot::PgSnapshotSource::new(pg_pool.clone()); + let snapshot_watch = tokio::time::timeout( + std::time::Duration::from_secs(60), + tokens::watch(snapshot_source).ready_owned(), + ) + .await + .context("timed out fetching the initial authorization snapshot")?; + // Start a logs sink into which agent loops may stream logs. let (logs_tx, logs_rx) = tokio::sync::mpsc::channel(8192); let logs_sink = control_plane_api::logs::serve_sink(pg_pool.clone(), logs_rx); @@ -348,20 +362,6 @@ async fn async_main(args: Args) -> Result<(), anyhow::Error> { } .shared(); - // Create the snapshot source and start the refresh loop. - // Snapshot fetches retry internally forever, so a persistent failure (a - // broken query, sops / KMS breakage) would otherwise hang here with the - // port unbound and nothing logged at error level. Bound the wait so that - // startup fails visibly, and fits within Cloud Run's 240s startup probe - // window even after the database retry budget above. - let snapshot_source = control_plane_api::snapshot::PgSnapshotSource::new(pg_pool.clone()); - let snapshot_watch = tokio::time::timeout( - std::time::Duration::from_secs(60), - tokens::watch(snapshot_source).ready_owned(), - ) - .await - .context("timed out fetching the initial authorization snapshot")?; - let controller_publication_cooldown = chrono::Duration::from_std(args.controller_publication_cooldown)?; let alert_config_defaults = args.controller_config.alert_config_defaults(); @@ -390,7 +390,7 @@ async fn async_main(args: Args) -> Result<(), anyhow::Error> { jwt_secret.as_bytes(), pg_pool.clone(), publisher.clone(), - snapshot_watch, + snapshot_watch.clone(), args.stripe_webhook_secret, )); let api_router = control_plane_api::build_router( @@ -417,12 +417,14 @@ async fn async_main(args: Args) -> Result<(), anyhow::Error> { .register(agent::publications::PublicationsExecutor { publisher, pg_pool: pg_pool.clone(), + snapshot_watch: snapshot_watch.clone(), runtime_v2_new_captures: args.runtime_v2_new_captures, runtime_v2_new_materializations: args.runtime_v2_new_materializations, runtime_v2_new_derivations: args.runtime_v2_new_derivations, }) .register(agent::DiscoverExecutor { handler: discover_handler, + snapshot_watch, }) .register(directive_executor) .register(connector_tags_executor) diff --git a/crates/agent/src/publications.rs b/crates/agent/src/publications.rs index 326bfd20296..4810786101b 100644 --- a/crates/agent/src/publications.rs +++ b/crates/agent/src/publications.rs @@ -1,4 +1,5 @@ use anyhow::Context; +use control_plane_api::Snapshot; use control_plane_api::publications::{Row, fetch_publication}; use models::draft_error; use tracing::info; @@ -15,6 +16,9 @@ use control_plane_api::{ pub struct PublicationsExecutor { pub publisher: Publisher, pub pg_pool: sqlx::PgPool, + /// Authorization Snapshot watch. Each poll pins one Snapshot from this + /// watch, which serves every authorization decision of the publication. + pub snapshot_watch: std::sync::Arc>, /// When true, newly-created captures are published onto runtime v2; see [`RuntimeV2Rollout`]. pub runtime_v2_new_captures: bool, /// When true, newly-created materializations are published onto runtime v2; see [`RuntimeV2Rollout`]. @@ -93,12 +97,17 @@ impl PublicationsExecutor { } } + // Pin one Snapshot for this poll: every authorization decision of the + // publication observes the same view. + let snapshot = self.snapshot_watch.token(); + let snapshot = snapshot.result().unwrap(); + let dry_run = row.dry_run; let draft_id = row.draft_id; let time_queued = chrono::Utc::now().signed_duration_since(row.updated_at); - let (status, draft_errors, final_pub_id) = match self.process(row).await { + let (status, draft_errors, final_pub_id) = match self.process(row, snapshot).await { Ok(result) => { if dry_run { specs::add_built_specs_to_draft_specs(draft_id, &result.built, &self.pg_pool) @@ -155,7 +164,7 @@ impl PublicationsExecutor { %row.dry_run, %row.user_id, ))] - async fn process(&self, row: Row) -> anyhow::Result { + async fn process(&self, row: Row, snapshot: &Snapshot) -> anyhow::Result { info!( %row.logs_token, %row.created_at, @@ -192,6 +201,13 @@ impl PublicationsExecutor { dry_run: row.dry_run, detail: row.detail.clone(), draft, + // `None` anchors authorization staleness to each spec's own + // `last_pub_id`, preserving the pre-Snapshot semantics where a + // denial is always evaluated against current-enough state. A + // follow-up anchors this to the queued publication row and defers + // on staleness instead. + started_at: None, + snapshot, verify_user_authz: true, default_data_plane_name: row.data_plane_name.clone().filter(|s| !s.is_empty()), initialize: ( @@ -201,7 +217,7 @@ impl PublicationsExecutor { new_derivations: self.runtime_v2_new_derivations, }, ExpandDraft { - filter_user_has_admin: true, + filter_user_authz: true, }, ), finalize: PruneUnboundCollections, diff --git a/crates/control-plane-api/src/discovers/mod.rs b/crates/control-plane-api/src/discovers/mod.rs index d5686059e49..f27ad368ab7 100644 --- a/crates/control-plane-api/src/discovers/mod.rs +++ b/crates/control-plane-api/src/discovers/mod.rs @@ -1,7 +1,7 @@ pub mod db; pub mod specs; -use crate::proxy_connectors::DiscoverConnectors; +use crate::{Snapshot, proxy_connectors::DiscoverConnectors}; use anyhow::Context; use models::discovers::{Changed, Changes}; @@ -14,7 +14,7 @@ pub use db::{Row, fetch_discover, resolve}; /// Represents the desire to discover an endpoint. The discovered bindings will be merged with /// those in the `base_model`. -pub struct Discover { +pub struct Discover<'a> { /// The name of the capture, which _must_ exist within the `draft`. pub capture_name: models::Capture, /// The data plane to use for the discover. For an existing capture, this @@ -41,6 +41,19 @@ pub struct Discover { /// from the live task's control-plane Id. Empty if the task doesn't exist /// yet: the connector assumes a current date for a new task's discover. pub created_at: String, + /// The authorization Snapshot pinned by the caller for this entire + /// operation: preflight checks, the connector RPC window, and the + /// post-RPC merge all consult this same instance, so one discover + /// observes exactly one authorization view regardless of refreshes + /// landing mid-flight. + pub snapshot: &'a crate::Snapshot, + /// Time at which the discover was queued, when the caller has a durable + /// one. Anchors authorization staleness of the merge's target collections: + /// a denial from a Snapshot older than this instant is provisional — + /// authority committed before queuing may be missing from it — and + /// reschedules the discover rather than silently dropping the live + /// collection. A Snapshot taken after this instant is authoritative. + pub started_at: Option, } #[derive(Debug)] @@ -161,7 +174,7 @@ impl DiscoverHandler { update_only = %req.update_only, image ))] - pub async fn discover(&self, db: &PgPool, req: Discover) -> anyhow::Result { + pub async fn discover(&self, db: &PgPool, req: Discover<'_>) -> anyhow::Result { let Discover { capture_name, data_plane, @@ -172,6 +185,8 @@ impl DiscoverHandler { reset_on_key_change, mut draft, created_at, + snapshot, + started_at, } = req; let Some(capture_def) = draft.captures.get_mut_by_key(&capture_name) else { @@ -232,6 +247,8 @@ impl DiscoverHandler { spec.resource_path_pointers, db, reset_on_key_change, + snapshot, + started_at, ) .await?; @@ -252,7 +269,7 @@ impl DiscoverHandler { async fn build_merged_catalog( capture_name: models::Capture, - user_id: uuid::Uuid, + user_id: Uuid, filter_user_authz: bool, update_only: bool, mut draft: tables::DraftCatalog, @@ -260,6 +277,8 @@ impl DiscoverHandler { resource_path_pointers: Vec, db: &PgPool, reset_on_key_change: bool, + snapshot: &Snapshot, + started_at: Option, ) -> anyhow::Result { let discovered_bindings = match specs::parse_response(discovered) .context("converting connector discovery response into specs") @@ -307,12 +326,13 @@ impl DiscoverHandler { .iter() .map(|b| b.target.to_string()) .collect::>(); - let live = crate::live_specs::get_live_specs( user_id, &collection_names, - filter_user_authz.then_some(models::Capability::Read), + filter_user_authz.then_some(models::authz::Capability::CatalogRead.into()), db, + snapshot, + started_at, ) .await?; diff --git a/crates/control-plane-api/src/fixtures/authz_specs.sql b/crates/control-plane-api/src/fixtures/authz_specs.sql new file mode 100644 index 00000000000..d96a7ea9ed5 --- /dev/null +++ b/crates/control-plane-api/src/fixtures/authz_specs.sql @@ -0,0 +1,88 @@ +-- Live specs with *deserializable* models and built specs, for tests that +-- actually load them into a `tables::LiveCatalog` (rather than only reading +-- their names or authorization). `alice.sql` deliberately stores `'{}'` specs, +-- which is enough for name-and-authorization tests but fails to deserialize. +-- +-- `carol` is admin of `carolCo/`; `dan` exists with no grants at all and so +-- models an unauthorized caller. +do $$ +declare + data_plane_one_id flowid := '111111111111'; + + carol_uid uuid := '33333333-3333-3333-3333-333333333333'; + dan_uid uuid := '44444444-4444-4444-4444-444444444444'; + + -- A flowid's high 41 bits are milliseconds since the Estuary epoch, and + -- authorization staleness is decided against that embedded timestamp. These + -- ids are therefore chosen to sit a few days *after* the epoch, so that a + -- Snapshot taken shortly before them is still comfortably after the zero id + -- that a not-yet-published spec resolves to. Spell them out in full: a + -- 12-hex-digit literal is widened to macaddr8 by inserting FF:FE in the + -- middle, which would scramble the timestamp. + collection_id flowid := '00:08:00:00:00:00:04:01'; + capture_id flowid := '00:08:00:00:00:00:04:02'; + materialization_id flowid := '00:08:00:00:00:00:04:03'; + last_pub_id flowid := '00:08:00:00:00:00:00:00'; + +begin + + insert into auth.users (id, email) values + (carol_uid, 'carol@example.com'), + (dan_uid, 'dan@example.com') + ; + -- Dan administers his own tenant but is granted nothing else — not even the + -- shared data-plane — so he is unauthorized to everything under `carolCo/`. + insert into public.user_grants (user_id, object_role, capability) values + (carol_uid, 'carolCo/', 'admin'), + (dan_uid, 'danCo/', 'admin') + ; + -- `carolCo/in/` may write to `carolCo/data/`; `carolCo/out/` is deliberately + -- granted nothing, so a spec under it fails its own read authorization. + insert into public.role_grants (subject_role, object_role, capability) values + ('carolCo/in/', 'carolCo/data/', 'write'), + ('carolCo/', 'ops/dp/public/', 'read') + ; + + perform internal.create_task(collection_id, 1::smallint, '00:00:00:00:00:00:00:00'::flowid); + perform internal.create_task(capture_id, 1::smallint, '00:00:00:00:00:00:00:00'::flowid); + perform internal.create_task(materialization_id, 1::smallint, '00:00:00:00:00:00:00:00'::flowid); + + insert into public.live_specs ( + id, controller_task_id, catalog_name, last_pub_id, spec_type, spec, built_spec, data_plane_id + ) values ( + collection_id, + collection_id, + 'carolCo/data/foo', + last_pub_id, + 'collection', + '{"schema": {"type": "object", "properties": {"id": {"type": "string"}}, "required": ["id"]}, "key": ["/id"]}', + '{"name": "carolCo/data/foo", "writeSchemaJson": "{}", "key": ["/id"], "partitionTemplate": {"name": "carolCo/data/foo/gen1234"}}', + data_plane_one_id + ), ( + capture_id, + capture_id, + 'carolCo/in/capture-foo', + last_pub_id, + 'capture', + '{"endpoint": {"connector": {"image": "source/test:test", "config": {}}}, "bindings": []}', + '{"name": "carolCo/in/capture-foo", "shardTemplate": {"id": "capture/carolCo/in/capture-foo/gen5678"}}', + data_plane_one_id + ), ( + materialization_id, + materialization_id, + 'carolCo/out/materialize-bar', + last_pub_id, + 'materialization', + '{"endpoint": {"connector": {"image": "materialize/test:test", "config": {}}}, "bindings": []}', + '{"name": "carolCo/out/materialize-bar", "shardTemplate": {"id": "materialization/carolCo/out/materialize-bar/gen9012"}}', + data_plane_one_id + ); + + -- The capture writes to the collection, which is what makes it reachable + -- from it via `fetch_expanded_live_specs`. + insert into public.live_spec_flows (source_id, target_id, flow_type) values + (capture_id, collection_id, 'capture') + ; + +end +$$; diff --git a/crates/control-plane-api/src/live_specs/db.rs b/crates/control-plane-api/src/live_specs/db.rs index 3215710aafc..7da76ae3d50 100644 --- a/crates/control-plane-api/src/live_specs/db.rs +++ b/crates/control-plane-api/src/live_specs/db.rs @@ -1,8 +1,7 @@ use crate::TextJson; -use models::{Capability, CatalogType, Id}; +use models::{CatalogType, Id}; use serde_json::value::RawValue; -use sqlx::types::{Json, Uuid}; -use tables::RoleGrant; +use sqlx::types::Json; /// Deletes the given live spec row, along with the corresponding `controller_jobs` row. pub async fn hard_delete_live_spec(id: Id, txn: &mut sqlx::PgConnection) -> sqlx::Result<()> { @@ -33,30 +32,18 @@ pub struct LiveSpec { pub spec: Option>>, pub built_spec: Option>>, pub inferred_schema_md5: Option, - // User's capability to the specification `catalog_name`. - pub user_capability: Option, - // Capabilities of the specification with respect to other roles. - pub spec_capabilities: Json>, pub dependency_hash: Option, } /// Returns a `LiveSpec` row for each of the given `names`. This will always return a row for each /// name, even if no live spec exists in the database. pub async fn fetch_live_specs( - user_id: Uuid, names: &[String], - fetch_user_capabilities: bool, - fetch_spec_capabilities: bool, db: impl sqlx::Executor<'_, Database = sqlx::Postgres>, ) -> sqlx::Result> { - // The materialized CTE here ensures that `user_roles` is only invoked once, - // and the results used for the rest of the query. - sqlx::query_as!( + let live_spec = sqlx::query_as!( LiveSpec, r#" - with user_roles as materialized ( - select role_prefix, capability from internal.user_roles($1) - ) select coalesce(ls.id, '00:00:00:00:00:00:00:00'::flowid) as "id!: Id", coalesce(ls.last_pub_id, '00:00:00:00:00:00:00:00'::flowid) as "last_pub_id!: Id", @@ -67,31 +54,16 @@ pub async fn fetch_live_specs( ls.spec as "spec: TextJson>", ls.built_spec as "built_spec: TextJson>", ls.inferred_schema_md5, - case when $3 then ( - select max(capability) from user_roles - where starts_with(names, user_roles.role_prefix) - ) else - null - end as "user_capability: Capability", - case when $4 then coalesce( - (select json_agg(row_to_json(role_grants)) - from role_grants - where starts_with(names, subject_role)), - '[]' - ) else - '[]' - end as "spec_capabilities!: Json>", ls.dependency_hash - from unnest($2::text[]) names + from unnest($1::text[]) names left outer join live_specs ls on ls.catalog_name = names "#, - user_id, names, - fetch_user_capabilities, - fetch_spec_capabilities, ) .fetch_all(db) - .await + .await?; + + Ok(live_spec) } pub struct InferredSchemaRow { @@ -122,7 +94,6 @@ pub async fn fetch_inferred_schemas( /// Queries for all non-deleted `live_specs` that are connected to the given `collection_names` via /// `live_spec_flows`. pub async fn fetch_expanded_live_specs( - user_id: Uuid, collection_names: &[&str], exclude_names: &[&str], db: impl sqlx::Executor<'_, Database = sqlx::Postgres>, @@ -132,7 +103,7 @@ pub async fn fetch_expanded_live_specs( r#" with collections(id) as ( select ls.id - from unnest($2::text[]) as names(catalog_name) + from unnest($1::text[]) as names(catalog_name) join live_specs ls on ls.catalog_name = names.catalog_name ), exp(id) as ( @@ -154,22 +125,11 @@ pub async fn fetch_expanded_live_specs( ls.spec as "spec: TextJson>", ls.built_spec as "built_spec: TextJson>", ls.inferred_schema_md5, - ( - select max(capability) from internal.user_roles($1) r - where starts_with(ls.catalog_name, r.role_prefix) - ) as "user_capability: Capability", - coalesce( - (select json_agg(row_to_json(role_grants)) - from role_grants - where starts_with(ls.catalog_name, subject_role)), - '[]' - ) as "spec_capabilities!: Json>", ls.dependency_hash from exp join live_specs ls on ls.id = exp.id - where ls.spec is not null and not ls.catalog_name = any($3); + where ls.spec is not null and not ls.catalog_name = any($2); "#, - user_id, collection_names as &[&str], exclude_names as &[&str], ) diff --git a/crates/control-plane-api/src/live_specs/mod.rs b/crates/control-plane-api/src/live_specs/mod.rs index c2849eb3288..a7863e2d366 100644 --- a/crates/control-plane-api/src/live_specs/mod.rs +++ b/crates/control-plane-api/src/live_specs/mod.rs @@ -1,39 +1,43 @@ mod db; use anyhow::Context; -use models::Capability; -use std::ops::Deref; -use uuid::Uuid; - pub use db::{ InferredSchemaRow, LiveSpec, fetch_expanded_live_specs, fetch_inferred_schemas, fetch_live_spec_names_by_prefix, fetch_live_specs, hard_delete_live_spec, }; +use std::ops::Deref; +use uuid::Uuid; /// Fetches live specs, returning them as a `tables::LiveCatalog`. Optionally /// filters the specs based on user capability. If `filter_capability` is /// `None`, then no filtering will be done. +/// +/// `started_at` anchors the staleness check to the given time (request-relative). +/// When `None`, staleness is anchored to each spec's publication time (spec-relative). +/// A denial from a snapshot older than the anchor is provisional — authority +/// committed before the anchor may be missing from it — and surfaces as a +/// retryable error. A snapshot taken after the anchor is authoritative: it is +/// guaranteed to reflect everything committed before the anchor, but not +/// necessarily changes committed after it. pub async fn get_live_specs( - user_id: Uuid, + user_id: uuid::Uuid, names: &[String], - filter_capability: Option, + filter_capability: Option, db: &sqlx::PgPool, + snapshot: &crate::Snapshot, + started_at: Option, ) -> anyhow::Result { let mut live = tables::LiveCatalog::default(); - // The query that's used by `fetch_live_specs` can be pretty slow because of how - // it queries authZ capabilities for each name, even if it doesn't exist. - // Limit each individual query to 512 names to avoid statement timeouts when - // fetching a large number of specs when `filter_capability` is `Some`. + // Fetch in batches of 512 names. The recursive per-name authorization + // work which originally motivated batching (see #1895) has moved + // in-process, but each returned row still carries unbounded `spec` and + // `built_spec` JSON documents, and a large discover can request thousands + // of names at once. Batching bounds each statement's execution and + // transfer time — keeping every statement clear of `statement_timeout` + // regardless of catalog size — at the cost of a round trip per batch. for names_chunk in names.chunks(512) { - let rows = db::fetch_live_specs( - user_id, - names_chunk, - filter_capability.is_some(), // fetch user capabilities only if needed - false, // we never need spec_capabilities here - db, - ) - .await?; + let rows = db::fetch_live_specs(names_chunk, db).await?; for row in rows { // Spec type might be null because we used to set it to null when deleting specs. // For recently deleted specs, it will still be present. @@ -44,14 +48,19 @@ pub async fn get_live_specs( continue; }; if let Some(min_capability) = filter_capability { - if !row - .user_capability - .is_some_and(|actual_capability| actual_capability >= min_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)? { continue; } } - let built_spec_json = row.built_spec.as_ref().ok_or_else(|| { + let built_spec_json: &Box = row.built_spec.as_ref().ok_or_else(|| { tracing::warn!(catalog_name = %row.catalog_name, id = %row.id, "got row with spec but not built_spec"); anyhow::anyhow!("missing built_spec for {:?}, but spec is non-null", row.catalog_name) })?.deref(); @@ -74,22 +83,34 @@ pub async fn get_live_specs( Ok(live) } +/// Fetches the live specs connected to `collection_names` — tasks that read +/// from or write to them — excluding `exclude_names`. When `filter_capability` +/// is set, specs to which the user lacks that capability are silently omitted: +/// expansion is filtering, so a denial is final regardless of Snapshot +/// freshness and is never surfaced as a retryable stale error. pub async fn get_connected_live_specs( user_id: Uuid, collection_names: &[&str], exclude_names: &[&str], - filter_capability: Option, + filter_capability: Option, db: &sqlx::PgPool, + snapshot: &crate::Snapshot, ) -> anyhow::Result { - let expanded_rows = - db::fetch_expanded_live_specs(user_id, collection_names, exclude_names, db).await?; + let expanded_rows = db::fetch_expanded_live_specs(collection_names, exclude_names, db).await?; let mut live = tables::LiveCatalog::default(); + for exp in expanded_rows { if let Some(minimum_capability) = filter_capability { - if !exp - .user_capability - .map(|c| c >= minimum_capability) - .unwrap_or(false) + // Expansion widens validation with specs the caller never named, so + // a denial is a final omission rather than an error, and never + // consults Snapshot freshness (`None` anchor): the worst case of a + // not-yet-observed grant is only a narrower validation, while an + // anchored check would defer nearly every publication touching a + // connected spec its user can't edit, since the pinned Snapshot + // almost always predates the queued row. + if !snapshot + .user_authorization(user_id, &exp.catalog_name, minimum_capability, None) + .ok_or_stale(&exp.catalog_name)? { continue; } @@ -122,3 +143,209 @@ pub async fn get_connected_live_specs( } Ok(live) } + +/// Both fetchers apply authorization in-process against a `Snapshot` rather than +/// in SQL, but they trust a denial differently. `get_live_specs` fetches specs +/// the caller explicitly named, where a wrongly-dropped spec corrupts the +/// operation's output; because the Snapshot lags Postgres, a denial is only +/// trusted once the Snapshot is authoritative for the operation asking — its +/// `started_at` request time when the caller has a durable one, or the denied +/// spec's own last publication otherwise — and until then the caller gets a +/// retryable `AuthorizationSnapshotStale` rather than a silently-dropped spec. +/// `get_connected_live_specs` expands to specs the caller never named, purely to +/// widen validation, so a denial is always a final silent omission and Snapshot +/// freshness is never consulted. +/// These tests pin the three-way outcome — included / dropped / retryable — for +/// the former, the exact instant the last two swap over, and the two-way +/// outcome for the latter. +#[cfg(test)] +mod tests { + use super::*; + + // From `fixtures/authz_specs.sql`. Carol is admin of `carolCo/`; Dan holds no + // grants at all 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"; + + /// 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. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_get_live_specs_unfiltered_never_stale(pool: sqlx::PgPool) { + let snapshot = stale(&pool).await; + let live = get_live_specs(DAN, &[COLLECTION.to_string()], None, &pool, &snapshot, None) + .await + .expect("an unfiltered fetch should not consult the Snapshot"); + + assert_eq!(1, live.collections.len()); + assert_eq!(COLLECTION, live.collections[0].collection.as_str()); + } + + /// An authorized caller gets the spec no matter how old the Snapshot is: + /// staleness only ever converts a *denial* into a retry. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_get_live_specs_authorized_is_included(pool: sqlx::PgPool) { + for snapshot in [stale(&pool).await, authoritative(&pool).await] { + let live = get_live_specs( + CAROL, + &[COLLECTION.to_string()], + Some(models::authz::Capability::CatalogRead.into()), + &pool, + &snapshot, + None, + ) + .await + .expect("carol is admin of carolCo/"); + + assert_eq!(1, live.collections.len()); + } + } + + /// An authoritative denial keeps the pre-existing behavior: the spec is + /// silently omitted rather than raising. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_get_live_specs_authoritative_denial_is_dropped(pool: sqlx::PgPool) { + let snapshot = authoritative(&pool).await; + let live = get_live_specs( + DAN, + &[COLLECTION.to_string()], + Some(models::authz::Capability::CatalogRead.into()), + &pool, + &snapshot, + None, + ) + .await + .expect("an authoritative denial is not an error"); + + assert!( + live.collections.is_empty(), + "an unauthorized spec should be omitted" + ); + } + + /// The new behavior: the same denial, judged by a Snapshot that predates the + /// spec, is retryable instead — the grant that would allow it may simply not + /// have propagated into this Snapshot yet. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_get_live_specs_stale_denial_is_retryable(pool: sqlx::PgPool) { + let snapshot = stale(&pool).await; + let err = get_live_specs( + DAN, + &[COLLECTION.to_string()], + Some(models::authz::Capability::CatalogRead.into()), + &pool, + &snapshot, + None, + ) + .await + .expect_err("a denial against a stale Snapshot should be retryable"); + + assert_stale_for(err, COLLECTION); + } + + /// `get_connected_live_specs` reaches specs by graph traversal rather than by + /// name, and it filters rather than authorizes: an unauthorized spec is + /// silently omitted, and the Snapshot's age never converts that omission + /// into a retryable error. The fixture's capture writes to the collection, + /// so it is reachable from it. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_get_connected_live_specs_filtering(pool: sqlx::PgPool) { + // Exclude the collection itself, leaving just the capture that writes it. + async fn connected( + pool: &sqlx::PgPool, + user: uuid::Uuid, + snapshot: &crate::Snapshot, + filter: Option, + ) -> anyhow::Result { + get_connected_live_specs(user, &[COLLECTION], &[COLLECTION], filter, pool, snapshot) + .await + } + let read_filter = Some(models::authz::CapabilitySet::from( + models::authz::Capability::CatalogRead, + )); + + let live = connected(&pool, CAROL, &authoritative(&pool).await, read_filter) + .await + .expect("carol is authorized"); + assert_eq!(1, live.captures.len()); + assert_eq!(CAPTURE, live.captures[0].capture.as_str()); + + let live = connected(&pool, DAN, &authoritative(&pool).await, read_filter) + .await + .expect("an authoritative denial is not an error"); + assert!(live.captures.is_empty()); + + let live = connected(&pool, DAN, &stale(&pool).await, read_filter) + .await + .expect("a denial filters silently even under a stale Snapshot"); + assert!(live.captures.is_empty()); + + let live = connected(&pool, DAN, &stale(&pool).await, None) + .await + .expect("an unfiltered traversal should not consult the Snapshot"); + assert_eq!(1, live.captures.len()); + } +} diff --git a/crates/control-plane-api/src/publications/initialize.rs b/crates/control-plane-api/src/publications/initialize.rs index b2648fd8180..8eb38a6b69e 100644 --- a/crates/control-plane-api/src/publications/initialize.rs +++ b/crates/control-plane-api/src/publications/initialize.rs @@ -1,16 +1,20 @@ use anyhow::Context; use itertools::Itertools; -use models::Capability; use std::future::Future; use uuid::Uuid; /// Initialize a draft prior to build/validation. This may add additional specs to the draft. +/// +/// `snapshot` is the publication's pinned authorization view; it must be the +/// same Snapshot the subsequent build uses, so that expansion and resolution +/// cannot disagree about one publication. pub trait Initialize: Send + Sync { fn initialize( &self, db: &sqlx::PgPool, user_id: Uuid, draft: &mut tables::DraftCatalog, + snapshot: &crate::Snapshot, ) -> impl Future> + Send; } @@ -22,6 +26,7 @@ impl Initialize for NoopInitialize { _db: &sqlx::PgPool, _user_id: Uuid, _draft: &mut tables::DraftCatalog, + _snapshot: &crate::Snapshot, ) -> anyhow::Result<()> { Ok(()) } @@ -37,20 +42,22 @@ where db: &sqlx::PgPool, user_id: Uuid, draft: &mut tables::DraftCatalog, + snapshot: &crate::Snapshot, ) -> anyhow::Result<()> { - self.0.initialize(db, user_id, draft).await?; - self.1.initialize(db, user_id, draft).await?; + self.0.initialize(db, user_id, draft, snapshot).await?; + self.1.initialize(db, user_id, draft, snapshot).await?; Ok(()) } } /// An `Initialize` that expands the draft to touch live specs that read from or write to /// any drafted collections. This may optionally filter the specs based on whether the user -/// has `admin` capability to them. +/// is authorized to edit them. pub struct ExpandDraft { - /// Whether to filter specs based on the user's capability. If true, then only specs for which - /// the user has `admin` capability will be added to the draft. - pub filter_user_has_admin: bool, + /// Whether to filter specs based on the user's capability. If true, then only specs for + /// which the user holds `SpecEdit` will be added to the draft — matching the capability + /// which publication requires of every drafted spec. + pub filter_user_authz: bool, } impl Initialize for ExpandDraft { @@ -58,13 +65,14 @@ impl Initialize for ExpandDraft { level = "debug", skip_all, err, - fields(filter_user_has_admin = self.filter_user_has_admin) + fields(filter_user_authz = self.filter_user_authz) )] async fn initialize( &self, db: &sqlx::PgPool, user_id: Uuid, draft: &mut tables::DraftCatalog, + snapshot: &crate::Snapshot, ) -> anyhow::Result<()> { // Expand the set of drafted specs to include any tasks that read from or write to any of // the published collections. We do this so that validation can catch any inconsistencies @@ -76,17 +84,16 @@ impl Initialize for ExpandDraft { .collect::>(); let all_drafted_specs = draft.all_spec_names().collect::>(); - let capability_filter = if self.filter_user_has_admin { - Some(Capability::Admin) - } else { - None - }; + let capability_filter = self + .filter_user_authz + .then_some(models::authz::Capability::SpecEdit.into()); let expanded_catalog = crate::live_specs::get_connected_live_specs( user_id, &drafted_collections, &all_drafted_specs, capability_filter, db, + snapshot, ) .await?; tracing::debug!( @@ -116,6 +123,7 @@ impl Initialize for RuntimeV2Rollout { db: &sqlx::PgPool, _user_id: Uuid, draft: &mut tables::DraftCatalog, + _snapshot: &crate::Snapshot, ) -> anyhow::Result<()> { let flag = models::Token::new(models::ENABLE_RUNTIME_V2); diff --git a/crates/control-plane-api/src/publications/mod.rs b/crates/control-plane-api/src/publications/mod.rs index a7bc7056eaf..05da666911e 100644 --- a/crates/control-plane-api/src/publications/mod.rs +++ b/crates/control-plane-api/src/publications/mod.rs @@ -1,11 +1,11 @@ -use std::u32; - use super::logs; +use crate::Snapshot; use anyhow::Context; use chrono::{DateTime, Utc}; use rand::Rng; use sqlx::Executor; use sqlx::types::Uuid; +use std::u32; use tables::BuiltRow; pub mod builds; @@ -29,7 +29,13 @@ use models::draft_error; /// Represents a desire to publish the given `draft`, along with associated metadata and behavior /// for handling draft initialization, build finalizing, and retrying failures. -pub struct DraftPublication { +pub struct DraftPublication< + 's, + Init: Initialize, + Fin: FinalizeBuild, + Ret: RetryPolicy, + C: WithCommit, +> { /// The id of the user that is publishing the draft. pub user_id: Uuid, /// Write logs to `internal.log_lines` using this token. @@ -42,8 +48,25 @@ pub struct DraftPublication, - /// Whether to check user permissions when publishing specs. If this is false, then all - /// permission checks will be skipped, and the publication may modify any specs. + /// The instant this publication was queued, which decides whether an + /// authorization denial is terminal or merely not-yet-observed by the + /// snapshot: a denial counts only once the snapshot was taken after it. + /// + /// This is distinct from `UncommittedBuild::started_at`, which is stamped + /// per build attempt. It must be durable across attempts for the retry to + /// converge, so it comes from the queued `publications` row (`updated_at`). + /// `None` means "no durable instant" — see [`specs::resolve_live_specs`]. + pub started_at: Option, + /// The authorization Snapshot to evaluate this publication against. One + /// pinned Snapshot serves every phase and internal retry of this + /// publication; its freshness relative to `started_at` decides whether a + /// denial is terminal or retryable (see [`specs::resolve_live_specs`]). + pub snapshot: &'s Snapshot, + /// Whether to verify that `user_id` is authorized to the drafted and + /// referenced catalog names, and to the selected data plane. Set `false` + /// by system-initiated publications (controllers, data-plane creation) + /// which are pre-authorized and may touch any spec. Spec-to-spec + /// `RoleGrant` checks are enforced regardless of this setting. pub verify_user_authz: bool, /// Default data plane to use for publishing new specs. This is optional only when the /// publication _only_ updates and/or deletes existing live specs. @@ -141,7 +164,7 @@ impl PublicationResult { } /// A PublishHandler is a Handler which publishes catalog specifications. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct Publisher { flowctl_go: std::path::PathBuf, builds_root: url::Url, @@ -270,16 +293,28 @@ impl Publisher { ))] pub async fn publish( &self, - publication: DraftPublication, + publication: DraftPublication<'_, Ini, Fin, Ret, C>, ) -> anyhow::Result { let mut retry_count = 0u32; loop { // Generate a new id on each attempt, so that we can retry `PublicationSuperseded` // errors with a greater id. let publication_id = self.next_id(); - let result = self + let result = match self .try_publish(publication_id, retry_count, &publication) - .await?; + .await + { + Ok(result) => result, + Err(err) if validation::is_authz_snapshot_stale(&err) => { + // The draft was denied by a Snapshot older than the denial + // freshness anchor, so the required grant may not be reflected + // yet. Request an early refresh and return the retryable error; + // task-based callers reschedule it against a newer Snapshot. + publication.snapshot.revoke.cancel(); + return Err(err); + } + Err(err) => return Err(err), + }; if result.status.is_success() || result.status.is_empty_draft() { return Ok(result); @@ -303,16 +338,19 @@ impl Publisher { draft: raw_draft, verify_user_authz, detail, + started_at, + snapshot, default_data_plane_name, initialize, finalize, retry: _, with_commit, - }: &DraftPublication, + }: &DraftPublication<'_, Ini, Fin, Ret, C>, ) -> anyhow::Result { let mut draft = raw_draft.clone_specs(); + let snapshot = *snapshot; initialize - .initialize(&self.db, *user_id, &mut draft) + .initialize(&self.db, *user_id, &mut draft, snapshot) .await .context("initializing draft")?; // It's important that we generate the pub id inside the retry loop so that we can @@ -327,6 +365,8 @@ impl Publisher { default_data_plane_name.as_deref(), *verify_user_authz, retry_count, + snapshot, + *started_at, ) .await?; finalize.finalize(&mut built).context("finalizing build")?; @@ -350,7 +390,7 @@ impl Publisher { /// Build and verify the given draft. This is `pub` only because we have existing tests that /// use it. If you want to publish something, use the `Publisher::publish` function instead. - #[tracing::instrument(level = "info", skip(self, draft))] + #[tracing::instrument(level = "info", skip(self, draft, snapshot))] pub async fn build( &self, user_id: Uuid, @@ -361,6 +401,8 @@ impl Publisher { explicit_plane_name: Option<&str>, verify_user_authz: bool, retry_count: u32, + snapshot: &crate::Snapshot, + started_at: Option, ) -> anyhow::Result { let start_time = tokens::now(); let build_id = self.id_gen.lock().unwrap().next(); @@ -406,6 +448,8 @@ impl Publisher { &self.db, verify_user_authz, explicit_plane_name, + snapshot, + started_at, ) .await?; diff --git a/crates/control-plane-api/src/publications/specs.rs b/crates/control-plane-api/src/publications/specs.rs index 74387fd44a0..b0c27faae0f 100644 --- a/crates/control-plane-api/src/publications/specs.rs +++ b/crates/control-plane-api/src/publications/specs.rs @@ -727,12 +727,35 @@ pub fn get_ops_collection_names() -> BTreeSet { names } +/// Builds the retryable `AuthorizationSnapshotStale` error returned when an +/// authorization denial was evaluated against a snapshot that isn't yet +/// authoritative for the operation being denied. +/// Resolves the live specs which a draft drafts or references, authorizing each +/// against `snapshot`. +/// +/// `started` is the instant the publication was queued, and decides whether an +/// authorization denial is terminal or merely not-yet-observed: a denial is +/// authoritative only once `snapshot` was taken after it. It must therefore be +/// durable across retries — a value re-stamped per attempt (`now()`) can never +/// be overtaken by a snapshot, so denials would retry forever. +/// +/// `None` is for callers with no such durable instant: controllers and ad-hoc +/// system publications, which construct a fresh publication per attempt and +/// carry their own retry/backoff. They fall back to anchoring on each denied +/// spec's own last publication, which bounds the window in which grants could +/// have been committed alongside the spec. Named data planes have no equivalent +/// fallback timestamp, so their denials remain terminal omissions. +/// +/// `verify_user_authz` skips only user-to-catalog authorization. Specification +/// `RoleGrant` checks remain mandatory. pub async fn resolve_live_specs( - user_id: Uuid, + user_id: uuid::Uuid, draft: &tables::DraftCatalog, db: &sqlx::PgPool, verify_user_authz: bool, explicit_plane_name: Option<&str>, + snapshot: &crate::Snapshot, + started: Option, ) -> anyhow::Result { // We're expecting to get a row for catalog name that's either drafted or referenced // by a drafted spec, even if the live spec does not exist. In that case, the row will @@ -759,15 +782,9 @@ pub async fn resolve_live_specs( } } - let rows = crate::live_specs::fetch_live_specs( - user_id, - &all_spec_names, - verify_user_authz, - true, // always fetch spec capabilities - db, - ) - .await - .context("fetching live specs")?; + let rows = crate::live_specs::fetch_live_specs(&all_spec_names, db) + .await + .context("fetching live specs")?; // Check the user and spec authorizations. // Start by making an easy way to lookup whether each row was drafted or not. @@ -782,14 +799,28 @@ pub async fn resolve_live_specs( let catalog_name = spec_row.catalog_name.as_str(); let n_errors = live.errors.len(); + // Use the queued publication time when available; callers without one + // fall back to the last publication time of the spec. + let freshness_anchor = Some(started.unwrap_or_else(|| spec_row.last_pub_id.timestamp())); + if drafted_names.contains(catalog_name) { // Get the metadata about the draft spec that matches this catalog name. // This must exist in `draft`, otherwise `spec_meta` will panic. let (catalog_type, reads_from, writes_to) = spec_meta(draft, catalog_name); let scope = tables::synthetic_scope(catalog_type, catalog_name); - // If the spec is included in the draft, then the user must have admin capability to it. - if verify_user_authz && !matches!(spec_row.user_capability, Some(Capability::Admin)) { + // If the spec is included in the draft, then the user must be + // authorized to edit it. + if verify_user_authz + && !snapshot + .user_authorization( + user_id, + &spec_row.catalog_name, + models::authz::Capability::SpecEdit, + freshness_anchor, + ) + .ok_or_stale(catalog_name)? + { live.errors.push(tables::Error { scope: scope.clone(), error: anyhow::anyhow!( @@ -800,30 +831,44 @@ pub async fn resolve_live_specs( // of referenced collections. continue; } - // Spec authz must always be checked, even if we're not checking user authz + // Spec authz must always be checked, even if we're not checking user authz. + // These spec-to-spec checks stay on legacy capabilities: they must agree + // with the runtime's task authorization, which enforces the same legacy + // roles when the task actually reads or writes collection journals. for source in reads_from { - if !spec_row.spec_capabilities.iter().any(|c| { - source.starts_with(c.object_role.as_str()) && c.capability >= Capability::Read - }) { + if !snapshot + .role_authorization( + &spec_row.catalog_name, + &source, + Capability::Read, + freshness_anchor, + ) + .ok_or_stale(catalog_name)? + { live.errors.push(tables::Error { scope: scope.clone(), error: anyhow::anyhow!( "Specification '{catalog_name}' is not read-authorized to '{source}'.\nAvailable grants are: {}", - serde_json::to_string_pretty(&spec_row.spec_capabilities.0).unwrap(), + serde_json::to_string_pretty(&snapshot.spec_capabilities(&spec_row.catalog_name)).unwrap(), ), }); } } for target in writes_to { - if !spec_row.spec_capabilities.iter().any(|c| { - target.starts_with(c.object_role.as_str()) - && matches!(c.capability, Capability::Write | Capability::Admin) - }) { + if !snapshot + .role_authorization( + &spec_row.catalog_name, + &target, + Capability::Write, + freshness_anchor, + ) + .ok_or_stale(catalog_name)? + { live.errors.push(tables::Error { scope: scope.clone(), error: anyhow::anyhow!( "Specification is not write-authorized to '{target}'.\nAvailable grants are: {}", - serde_json::to_string_pretty(&spec_row.spec_capabilities.0).unwrap(), + serde_json::to_string_pretty(&snapshot.spec_capabilities(&spec_row.catalog_name)).unwrap(), ), }); } @@ -832,15 +877,19 @@ pub async fn resolve_live_specs( // access capability to them as long as they are not drafted. } else if !ops_collection_names.contains(&spec_row.catalog_name) { // This is a live spec that is not included in the draft. - // The user needs read capability to it because it was referenced by one of the specs - // in their draft. Note that the _user_ does not need `Capability::Write` as long as + // The user needs `CatalogRead` to it because it was referenced by one of the specs + // in their draft. Note that the _user_ does not need any write capability as long as // the _spec_ is authorized to do what it needs. The user just needs to be allowed to // know it exists. if verify_user_authz - && !spec_row - .user_capability - .map(|c| c >= Capability::Read) - .unwrap_or(false) + && !snapshot + .user_authorization( + user_id, + &spec_row.catalog_name, + models::authz::Capability::CatalogRead, + freshness_anchor, + ) + .ok_or_stale(catalog_name)? { let scope = tables::synthetic_scope("unauthorized", &spec_row.catalog_name); live.errors.push(tables::Error { @@ -911,7 +960,7 @@ pub async fn resolve_live_specs( // Fetch data planes that are referenced by live specs (`data_plane_ids`), // or by storage mappings (`data_plane_names`), or by `explicit_plane_name`. - let data_plane_names: Vec<&str> = live + let candidate_data_plane_names: Vec<&str> = live .storage_mappings .iter() .flat_map(|m| m.data_planes.iter().map(String::as_str)) @@ -920,6 +969,21 @@ pub async fn resolve_live_specs( .dedup() .collect(); + let mut data_plane_names = Vec::with_capacity(candidate_data_plane_names.len()); + for name in candidate_data_plane_names { + // Legacy `read` is what conveys deploy-level trust in a data plane; + // no narrow capability bit expresses that trust yet. + if !verify_user_authz + || snapshot + .user_authorization(user_id, name, models::Capability::Read, started) + .ok_or_stale(name)? + { + data_plane_names.push(name); + } + } + + // IDs preserve the assignments of live specs already accepted above. They + // are not user-selected plane names and intentionally bypass this user check. data_plane_ids.sort(); data_plane_ids.dedup(); @@ -932,14 +996,10 @@ pub async fn resolve_live_specs( FROM UNNEST($1::flowid[]) AS t(id) ), data_plane_names AS ( + -- Names have already passed the caller's user-authorization policy, + -- so no in-SQL authorization check is needed here. SELECT name FROM UNNEST($2::text[]) AS t(name) - -- User must be read-authorized to data-plane. - WHERE EXISTS ( - SELECT 1 - FROM internal.user_roles($3, 'read') AS r - WHERE starts_with(t.name, r.role_prefix) - ) ) SELECT d.id AS "control_id: Id", @@ -961,7 +1021,6 @@ pub async fn resolve_live_specs( "#, &data_plane_ids as &[Id], &data_plane_names as &[&str], - user_id as Uuid, ) .fetch_all(db) .await? @@ -1151,3 +1210,367 @@ mod test { } } } + +/// `resolve_live_specs` makes four independent authorization decisions per row — +/// the drafter must hold `SpecEdit` to a drafted spec; a drafted spec must itself +/// be read-authorized to each source and write-authorized to each target; and the +/// user must hold `CatalogRead` to any *referenced* spec. Named data planes add +/// another user-authorization decision. Each denial is evaluated against a `Snapshot` +/// and short-circuits with retryable `AuthorizationSnapshotStale` when that +/// Snapshot is not authoritative for the operation. +/// +/// These tests pin both halves of every branch: what a stale Snapshot returns, +/// and the (unchanged) error text an authoritative one reports. +#[cfg(test)] +mod resolve_tests { + use super::*; + + // From `fixtures/authz_specs.sql`. + 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 MATERIALIZATION: &str = "carolCo/out/materialize-bar"; + const PLANE: &str = "ops/dp/public/aws-us-west-2-c1"; + + fn draft_of(catalog_json: serde_json::Value) -> tables::DraftCatalog { + let catalog: models::Catalog = + serde_json::from_value(catalog_json).expect("failed to parse catalog"); + tables::DraftCatalog::from(catalog) + } + + /// A materialization drafted under `carolCo/out/`, which holds no grants and + /// so is not read-authorized to `sources`. + fn materialization_draft(sources: &[&str]) -> tables::DraftCatalog { + draft_of(serde_json::json!({ + "materializations": { + MATERIALIZATION: { + "endpoint": { "connector": { "image": "materialize/test:test", "config": {} } }, + "bindings": sources.iter().map(|source| serde_json::json!({ + "resource": { "table": "t" }, + "source": source, + })).collect::>(), + } + } + })) + } + + /// A capture drafted under `carolCo/in/`, which may write to `carolCo/data/` + /// but nowhere else. + fn capture_draft(targets: &[&str]) -> tables::DraftCatalog { + draft_of(serde_json::json!({ + "captures": { + CAPTURE: { + "endpoint": { "connector": { "image": "source/test:test", "config": {} } }, + "bindings": targets.iter().map(|target| serde_json::json!({ + "resource": { "id": "r" }, + "target": target, + })).collect::>(), + } + } + })) + } + + /// 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 + .iter() + .map(|e| (e.scope.to_string(), format!("{:#}", e.error))) + .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 hold `SpecEdit` to it. + /// Dan does not, but the denial is only definitive once the Snapshot + /// outlives the spec. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_drafted_spec_requires_spec_edit(pool: sqlx::PgPool) { + let draft = draft_of(serde_json::json!({ + "collections": { + COLLECTION: { + "schema": { "type": "object", "properties": { "id": { "type": "string" } } }, + "key": ["/id"] + } + } + })); + + let err = resolve_live_specs(DAN, &draft, &pool, true, None, &stale(&pool).await, None) + .await + .expect_err("a denial against a stale Snapshot should be retryable"); + assert_stale_for(err, COLLECTION); + + let live = resolve_live_specs( + DAN, + &draft, + &pool, + true, + None, + &authoritative(&pool).await, + None, + ) + .await + .expect("an authoritative denial is reported, not raised"); + insta::assert_debug_snapshot!(error_pairs(&live), @r#" + [ + ( + "flow://collection/carolCo/data/foo", + "User is not authorized to create or change this catalog name", + ), + ] + "#); + } + + /// Branch 2: a drafted spec must itself be read-authorized to each source. + /// Carol admins the whole tenant, so the user check passes and only the + /// *spec's* own role grants are at issue. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_drafted_spec_reads_from_authz(pool: sqlx::PgPool) { + let draft = materialization_draft(&[COLLECTION]); + + let err = resolve_live_specs(CAROL, &draft, &pool, true, None, &stale(&pool).await, None) + .await + .expect_err("a denial against a stale Snapshot should be retryable"); + assert_stale_for(err, MATERIALIZATION); + + let live = resolve_live_specs( + CAROL, + &draft, + &pool, + true, + None, + &authoritative(&pool).await, + None, + ) + .await + .expect("an authoritative denial is reported, not raised"); + // The rendered grant list comes from `Snapshot::spec_capabilities`, which + // replaced a SQL-computed column; pin it so the two can't drift. + insta::assert_debug_snapshot!(error_pairs(&live), @r#" + [ + ( + "flow://materialization/carolCo/out/materialize-bar", + "Specification 'carolCo/out/materialize-bar' is not read-authorized to 'carolCo/data/foo'.\nAvailable grants are: [\n {\n \"subject_role\": \"carolCo/\",\n \"object_role\": \"ops/dp/public/\",\n \"capability\": \"read\",\n \"bundles\": []\n }\n]", + ), + ] + "#); + } + + /// Branch 3: a drafted spec must be write-authorized to each target. + /// `carolCo/in/` may write to `carolCo/data/` but nowhere else. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_drafted_spec_writes_to_authz(pool: sqlx::PgPool) { + let draft = capture_draft(&["carolCo/elsewhere/thing"]); + + let err = resolve_live_specs(CAROL, &draft, &pool, true, None, &stale(&pool).await, None) + .await + .expect_err("a denial against a stale Snapshot should be retryable"); + assert_stale_for(err, CAPTURE); + + let live = resolve_live_specs( + CAROL, + &draft, + &pool, + true, + None, + &authoritative(&pool).await, + None, + ) + .await + .expect("an authoritative denial is reported, not raised"); + insta::assert_debug_snapshot!(error_pairs(&live), @r#" + [ + ( + "flow://capture/carolCo/in/capture-foo", + "Specification is not write-authorized to 'carolCo/elsewhere/thing'.\nAvailable grants are: [\n {\n \"subject_role\": \"carolCo/\",\n \"object_role\": \"ops/dp/public/\",\n \"capability\": \"read\",\n \"bundles\": []\n },\n {\n \"subject_role\": \"carolCo/in/\",\n \"object_role\": \"carolCo/data/\",\n \"capability\": \"write\",\n \"bundles\": []\n }\n]", + ), + ] + "#); + } + + /// The write-authorized target resolves cleanly, confirming the branch above + /// fails for the reason claimed rather than incidentally. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_authorized_draft_resolves_without_errors(pool: sqlx::PgPool) { + let draft = capture_draft(&[COLLECTION]); + + for snapshot in [stale(&pool).await, authoritative(&pool).await] { + let live = resolve_live_specs(CAROL, &draft, &pool, true, Some(PLANE), &snapshot, None) + .await + .expect("an authorized draft resolves"); + + assert!( + live.errors.is_empty(), + "unexpected errors: {:?}", + error_pairs(&live) + ); + assert_eq!(1, live.captures.len()); + assert_eq!(1, live.collections.len()); + assert_eq!( + vec![PLANE], + live.data_planes + .iter() + .map(|d| d.data_plane_name.as_str()) + .collect::>(), + ); + } + } + + /// Branch 4: a *referenced* (non-drafted) spec only requires `CatalogRead`. + /// Dan admins `danCo/`, so his own drafted spec passes, and the denial lands + /// on `carolCo/data/foo` — which, being an existing spec, can be stale. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_referenced_spec_requires_catalog_read(pool: sqlx::PgPool) { + let draft = draft_of(serde_json::json!({ + "materializations": { + "danCo/materialize-x": { + "endpoint": { "connector": { "image": "materialize/test:test", "config": {} } }, + "bindings": [ { "resource": { "table": "t" }, "source": COLLECTION } ], + } + } + })); + + let err = resolve_live_specs(DAN, &draft, &pool, true, None, &stale(&pool).await, None) + .await + .expect_err("a denial against a stale Snapshot should be retryable"); + assert_stale_for(err, COLLECTION); + + let live = resolve_live_specs( + DAN, + &draft, + &pool, + true, + None, + &authoritative(&pool).await, + None, + ) + .await + .expect("an authoritative denial is reported, not raised"); + insta::assert_debug_snapshot!(error_pairs(&live), @r#" + [ + ( + "flow://unauthorized/carolCo/data/foo", + "User is not authorized to read this catalog name", + ), + ( + "flow://materialization/danCo/materialize-x", + "Specification 'danCo/materialize-x' is not read-authorized to 'carolCo/data/foo'.\nAvailable grants are: []", + ), + ] + "#); + } + + /// Without a durable request timestamp (`started: None`), a brand-new spec + /// falls back to its zero `last_pub_id` as the freshness anchor, so its + /// denial is terminal against any Snapshot — this keeps such a first + /// publication from looping instead of reporting its error. This holds + /// only for the `None` fallback: a queued publication supplies `started`, + /// which replaces the anchor and can make the same denial retryable. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_new_spec_denial_without_started_is_terminal(pool: sqlx::PgPool) { + let draft = draft_of(serde_json::json!({ + "collections": { + "carolCo/data/brand-new": { + "schema": { "type": "object", "properties": { "id": { "type": "string" } } }, + "key": ["/id"] + } + } + })); + + let live = resolve_live_specs(DAN, &draft, &pool, true, None, &stale(&pool).await, None) + .await + .expect("without a request anchor, a spec with no publication history cannot be stale"); + insta::assert_debug_snapshot!(error_pairs(&live), @r#" + [ + ( + "flow://collection/carolCo/data/brand-new", + "User is not authorized to create or change this catalog name", + ), + ] + "#); + } + + /// Spec-level (`reads_from` / `writes_to`) checks remain active when user + /// authorization is skipped. This test pins that stale denials from those + /// checks still propagate as retryable errors. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures(path = "../fixtures", scripts("data_planes", "authz_specs")) + )] + async fn test_spec_authz_staleness_applies_without_user_authz(pool: sqlx::PgPool) { + let draft = capture_draft(&["carolCo/elsewhere/thing"]); + + let err = resolve_live_specs( + uuid::Uuid::nil(), + &draft, + &pool, + false, // verify_user_authz + None, + &stale(&pool).await, + None, + ) + .await + .expect_err("spec authorization is checked regardless of verify_user_authz"); + assert_stale_for(err, CAPTURE); + } +} diff --git a/crates/control-plane-api/src/server/create_data_plane.rs b/crates/control-plane-api/src/server/create_data_plane.rs index 15ca92f9802..cacdd507887 100644 --- a/crates/control-plane-api/src/server/create_data_plane.rs +++ b/crates/control-plane-api/src/server/create_data_plane.rs @@ -232,12 +232,18 @@ 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, draft, dry_run: false, 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"), // We've already validated that the user can admin `ops/`, // so further authZ checks are unnecessary. verify_user_authz: false, diff --git a/crates/control-plane-api/src/server/update_l2_reporting.rs b/crates/control-plane-api/src/server/update_l2_reporting.rs index 9bf0e42d082..f7fc33ec5a0 100644 --- a/crates/control-plane-api/src/server/update_l2_reporting.rs +++ b/crates/control-plane-api/src/server/update_l2_reporting.rs @@ -295,12 +295,18 @@ 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, draft, dry_run, 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"), default_data_plane_name: if default_data_plane.trim().is_empty() { None } else {