From b2de4b89d160a8d31ad0f317ebe450416ecf652e Mon Sep 17 00:00:00 2001 From: Brian Bartman Date: Wed, 12 Aug 2026 11:40:37 +0000 Subject: [PATCH] control-plane: centralize spec-fetch authorization as Snapshot::spec_fetch_authorization The spec-fetch policy shared by get_live_specs and evolutions-style named fetches is centralized as Snapshot::spec_fetch_authorization so enforcement points cannot drift: staleness anchors on the operation's durable queued instant when the caller has one, and otherwise on the spec's own last publication. Authoritative denials drop the spec (pre-existing behavior); provisional denials surface as the retryable AuthorizationSnapshotStale error. The connected-spec expansion path deliberately does NOT adopt this policy: expansion widens validation with specs the caller never named, so a denial there remains a final omission with an unanchored check. Per-module test Snapshot helpers consolidate into a shared test_support module, and the dead ExpandedRow struct is removed from db_complete. --- crates/control-plane-api/src/lib.rs | 2 + .../control-plane-api/src/live_specs/mod.rs | 63 +++---------------- .../src/publications/db_complete.rs | 21 +------ .../src/publications/specs.rs | 45 +------------ .../control-plane-api/src/server/snapshot.rs | 24 +++++++ crates/control-plane-api/src/test_support.rs | 56 +++++++++++++++++ 6 files changed, 92 insertions(+), 119 deletions(-) create mode 100644 crates/control-plane-api/src/test_support.rs diff --git a/crates/control-plane-api/src/lib.rs b/crates/control-plane-api/src/lib.rs index ca16775c095..dab2210a9ed 100644 --- a/crates/control-plane-api/src/lib.rs +++ b/crates/control-plane-api/src/lib.rs @@ -24,6 +24,8 @@ mod text_json; #[cfg(test)] pub(crate) mod test_server; +#[cfg(test)] +pub(crate) mod test_support; /// TextJson encodes JSON for Postgres while preserving property ordering. pub use text_json::TextJson; diff --git a/crates/control-plane-api/src/live_specs/mod.rs b/crates/control-plane-api/src/live_specs/mod.rs index a7863e2d366..1576cc99ac3 100644 --- a/crates/control-plane-api/src/live_specs/mod.rs +++ b/crates/control-plane-api/src/live_specs/mod.rs @@ -48,15 +48,13 @@ pub async fn get_live_specs( continue; }; if let Some(min_capability) = filter_capability { - // For discovers, anchor to the discover request time (started_at). - // For other callers, anchor to the spec's publication time. - // An authoritative denial is today's silent drop; a provisional - // one surfaces as a retryable stale error. - let anchor = started_at.unwrap_or_else(|| row.last_pub_id.timestamp()); - if !snapshot - .user_authorization(user_id, &row.catalog_name, min_capability, Some(anchor)) - .ok_or_stale(&row.catalog_name)? - { + if !snapshot.spec_fetch_authorization( + user_id, + &row.catalog_name, + min_capability, + started_at, + row.last_pub_id, + )? { continue; } } @@ -161,6 +159,7 @@ pub async fn get_connected_live_specs( #[cfg(test)] mod tests { use super::*; + use crate::test_support::{assert_stale_for, authoritative, stale}; // From `fixtures/authz_specs.sql`. Carol is admin of `carolCo/`; Dan holds no // grants at all and so models an unauthorized caller. @@ -169,52 +168,6 @@ mod tests { const COLLECTION: &str = "carolCo/data/foo"; const CAPTURE: &str = "carolCo/in/capture-foo"; - /// Staleness compares the Snapshot's `taken` against the timestamp embedded - /// in a spec's `last_pub_id`, so read that back rather than recomputing it — - /// `flowid` is `macaddr8`, which silently widens short literals. - async fn published_at(pool: &sqlx::PgPool) -> tokens::DateTime { - sqlx::query_scalar!( - r#"select last_pub_id as "last_pub_id: models::Id" - from live_specs where catalog_name = $1"#, - COLLECTION, - ) - .fetch_one(pool) - .await - .expect("fixture collection should exist") - .timestamp() - } - - /// A Snapshot holding the fixture's real grants, stamped `offset` away from - /// the instant the fixture's specs were published. - async fn snapshot_offset(pool: &sqlx::PgPool, offset: chrono::TimeDelta) -> crate::Snapshot { - let mut decrypted_hmac_keys = std::collections::HashMap::new(); - let data = crate::snapshot::try_fetch(pool, &mut decrypted_hmac_keys) - .await - .expect("failed to fetch snapshot"); - crate::Snapshot::new(published_at(pool).await + offset, data) - } - - /// Taken clear of the publication plus `TEMPORAL_SKEW`: denials are definitive. - async fn authoritative(pool: &sqlx::PgPool) -> crate::Snapshot { - snapshot_offset(pool, crate::Snapshot::TEMPORAL_SKEW * 4).await - } - - /// Taken before the publication it would judge: denials are retryable. - async fn stale(pool: &sqlx::PgPool) -> crate::Snapshot { - snapshot_offset(pool, -crate::Snapshot::TEMPORAL_SKEW * 4).await - } - - fn assert_stale_for(err: anyhow::Error, catalog_name: &str) { - assert!( - validation::is_authz_snapshot_stale(&err), - "expected a retryable stale-snapshot error, got: {err:#}" - ); - assert!( - err.to_string().contains(catalog_name), - "stale error should name the offending spec, got: {err:#}" - ); - } - /// With no capability filter the Snapshot is never consulted, so even a /// wholly unauthorized caller reading against a stale Snapshot gets the spec. /// This is the path controllers and other system callers take. diff --git a/crates/control-plane-api/src/publications/db_complete.rs b/crates/control-plane-api/src/publications/db_complete.rs index a00f19c0d09..e9f92e549c4 100644 --- a/crates/control-plane-api/src/publications/db_complete.rs +++ b/crates/control-plane-api/src/publications/db_complete.rs @@ -1,6 +1,6 @@ use crate::FlowType; -use super::{Capability, CatalogType, Id, TextJson as Json}; +use super::{CatalogType, Id, TextJson as Json}; use chrono::prelude::*; use serde::Serialize; @@ -449,25 +449,6 @@ pub async fn find_tenant_quotas( .await } -#[derive(Debug)] -pub struct ExpandedRow { - // Name of the specification. - pub catalog_name: String, - // Last build ID of the live spec. - pub last_build_id: Id, - // Last publication ID of the live spec. - pub last_pub_id: Id, - // Current live specification of this expansion. - // It won't be changed by this publication. - pub live_spec: Json>, - // ID of the expanded live specification. - pub live_spec_id: Id, - // Spec type of the live specification. - pub live_type: CatalogType, - // User's capability to the specification `catalog_name`. - pub user_capability: Option, -} - pub async fn delete_stale_flow( live_spec_id: Id, catalog_type: CatalogType, diff --git a/crates/control-plane-api/src/publications/specs.rs b/crates/control-plane-api/src/publications/specs.rs index 8252cfecf2f..84e932c8529 100644 --- a/crates/control-plane-api/src/publications/specs.rs +++ b/crates/control-plane-api/src/publications/specs.rs @@ -1224,6 +1224,7 @@ mod test { #[cfg(test)] mod resolve_tests { use super::*; + use crate::test_support::{assert_stale_for, authoritative, published_at, stale}; // From `fixtures/authz_specs.sql`. const CAROL: uuid::Uuid = uuid::uuid!("33333333-3333-3333-3333-333333333333"); @@ -1274,39 +1275,6 @@ mod resolve_tests { })) } - /// Staleness compares the Snapshot's `taken` against the timestamp embedded - /// in a spec's `last_pub_id`, so read that back rather than recomputing it — - /// `flowid` is `macaddr8`, which silently widens short literals. - async fn published_at(pool: &sqlx::PgPool) -> tokens::DateTime { - sqlx::query_scalar!( - r#"select last_pub_id as "last_pub_id: models::Id" - from live_specs where catalog_name = $1"#, - COLLECTION, - ) - .fetch_one(pool) - .await - .expect("fixture collection should exist") - .timestamp() - } - - async fn snapshot_offset(pool: &sqlx::PgPool, offset: chrono::TimeDelta) -> crate::Snapshot { - let mut decrypted_hmac_keys = std::collections::HashMap::new(); - let data = crate::snapshot::try_fetch(pool, &mut decrypted_hmac_keys) - .await - .expect("failed to fetch snapshot"); - crate::Snapshot::new(published_at(pool).await + offset, data) - } - - /// Taken clear of the publication plus `TEMPORAL_SKEW`: denials are definitive. - async fn authoritative(pool: &sqlx::PgPool) -> crate::Snapshot { - snapshot_offset(pool, crate::Snapshot::TEMPORAL_SKEW * 4).await - } - - /// Taken before the publication it would judge: denials are retryable. - async fn stale(pool: &sqlx::PgPool) -> crate::Snapshot { - snapshot_offset(pool, -crate::Snapshot::TEMPORAL_SKEW * 4).await - } - /// Renders `live.errors` as `(scope, message)` pairs for snapshot assertions. fn error_pairs(live: &tables::LiveCatalog) -> Vec<(String, String)> { live.errors @@ -1315,17 +1283,6 @@ mod resolve_tests { .collect() } - fn assert_stale_for(err: anyhow::Error, catalog_name: &str) { - assert!( - validation::is_authz_snapshot_stale(&err), - "expected a retryable stale-snapshot error, got: {err:#}" - ); - assert!( - err.to_string().contains(catalog_name), - "stale error should name the offending spec, got: {err:#}" - ); - } - /// Branch 1: a user drafting an existing spec must hold `SpecEdit` to it. /// Dan does not, but the denial is only definitive once the Snapshot /// outlives the spec. diff --git a/crates/control-plane-api/src/server/snapshot.rs b/crates/control-plane-api/src/server/snapshot.rs index 2281a60ae3e..8d210e69e88 100644 --- a/crates/control-plane-api/src/server/snapshot.rs +++ b/crates/control-plane-api/src/server/snapshot.rs @@ -258,6 +258,30 @@ impl Snapshot { ) } + /// Evaluate whether `user_id` may fetch the live spec `catalog_name` with + /// `capability`: the policy applied when fetching specs the caller + /// explicitly named (`live_specs::get_live_specs`). Staleness anchors on + /// `started` — the fetching operation's durable queued instant — when the + /// caller has one, and otherwise on the spec's own last publication, which + /// bounds the window in which grants could have been committed alongside + /// the spec. + /// + /// `Ok(false)` is an authoritative denial: callers drop or suppress the + /// spec, the pre-existing behavior. A provisional denial surfaces as the + /// retryable `AuthorizationSnapshotStale` error instead. + pub fn spec_fetch_authorization( + &self, + user_id: uuid::Uuid, + catalog_name: &str, + capability: impl Into, + started: Option, + last_pub_id: models::Id, + ) -> Result { + let anchor = started.unwrap_or_else(|| last_pub_id.timestamp()); + self.user_authorization(user_id, catalog_name, capability, Some(anchor)) + .ok_or_stale(catalog_name) + } + /// Evaluate whether `subject` (a catalog spec acting as a role) holds /// `capability` to `object` under this Snapshot's role grants, classified /// against `anchor` freshness (see `resolve_authorization`). diff --git a/crates/control-plane-api/src/test_support.rs b/crates/control-plane-api/src/test_support.rs new file mode 100644 index 00000000000..6358df779ad --- /dev/null +++ b/crates/control-plane-api/src/test_support.rs @@ -0,0 +1,56 @@ +//! Shared helpers for authorization tests over the `authz_specs` fixture: +//! Snapshots of the database's real grants, stamped relative to the fixture's +//! publication instant so each test chooses whether a denial is authoritative +//! or provisional. + +/// The `authz_specs.sql` collection whose publication anchors staleness. +const ANCHOR_COLLECTION: &str = "carolCo/data/foo"; + +/// Staleness compares the Snapshot's `taken` against the timestamp embedded +/// in a spec's `last_pub_id`, so read that back rather than recomputing it — +/// `flowid` is `macaddr8`, which silently widens short literals. +pub(crate) async fn published_at(pool: &sqlx::PgPool) -> tokens::DateTime { + sqlx::query_scalar::<_, models::Id>( + "select last_pub_id from live_specs where catalog_name = $1", + ) + .bind(ANCHOR_COLLECTION) + .fetch_one(pool) + .await + .expect("fixture collection should exist") + .timestamp() +} + +/// A Snapshot holding the fixture's real grants, stamped `offset` away from +/// the instant the fixture's specs were published. +pub(crate) async fn snapshot_offset( + pool: &sqlx::PgPool, + offset: chrono::TimeDelta, +) -> crate::Snapshot { + let mut decrypted_hmac_keys = std::collections::HashMap::new(); + let data = crate::snapshot::try_fetch(pool, &mut decrypted_hmac_keys) + .await + .expect("failed to fetch snapshot"); + crate::Snapshot::new(published_at(pool).await + offset, data) +} + +/// Taken clear of the publication plus `TEMPORAL_SKEW`: denials are definitive. +pub(crate) async fn authoritative(pool: &sqlx::PgPool) -> crate::Snapshot { + snapshot_offset(pool, crate::Snapshot::TEMPORAL_SKEW * 4).await +} + +/// Taken before the publication it would judge: denials are retryable. +pub(crate) async fn stale(pool: &sqlx::PgPool) -> crate::Snapshot { + snapshot_offset(pool, -crate::Snapshot::TEMPORAL_SKEW * 4).await +} + +/// Asserts `err` is the retryable stale-snapshot error and names the spec. +pub(crate) fn assert_stale_for(err: anyhow::Error, catalog_name: &str) { + assert!( + validation::is_authz_snapshot_stale(&err), + "expected a retryable stale-snapshot error, got: {err:#}" + ); + assert!( + err.to_string().contains(catalog_name), + "stale error should name the offending spec, got: {err:#}" + ); +}