control-plane: centralize spec-fetch authorization as Snapshot::spec_fetch_authorization (5/7)

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;

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.

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<Box<RawValue>>,
-    // 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<Capability>,
-}
-
 pub async fn delete_stale_flow(
     live_spec_id: Id,
     catalog_type: CatalogType,

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.

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<models::authz::CapabilitySet>,
+        started: Option<tokens::DateTime>,
+        last_pub_id: models::Id,
+    ) -> Result<bool, validation::Error> {
+        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`).

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:#}"
+    );
+}