diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index a78cc34140f..47a9ca2f757 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -1589,7 +1589,7 @@ impl TestHarness { maybe_claims: control_plane_api::MaybeControlClaims::with_verified(verified), original_uri: axum::http::Uri::from_static("/graphql"), pg_pool: self.pool.clone(), - refresh: app.snapshot.token(), + refresh: app.snapshot_watch.token(), retry_after: tokens::DateTime::UNIX_EPOCH, started: tokens::now(), locale: control_plane_api::Locale::EnUS, diff --git a/crates/control-plane-api/src/envelope.rs b/crates/control-plane-api/src/envelope.rs index 3ca7d720771..ef6b28cc57f 100644 --- a/crates/control-plane-api/src/envelope.rs +++ b/crates/control-plane-api/src/envelope.rs @@ -272,7 +272,7 @@ impl axum::extract::FromRequestParts> for Envelope { Ok(Envelope { maybe_claims, retry_after: retry_after.unwrap_or(tokens::DateTime::UNIX_EPOCH), - refresh: state.snapshot.token(), + refresh: state.snapshot_watch.token(), started: started.unwrap_or_else(|| tokens::now()), pg_pool: state.pg_pool.clone(), original_uri, diff --git a/crates/control-plane-api/src/lib.rs b/crates/control-plane-api/src/lib.rs index 856766a693e..ca16775c095 100644 --- a/crates/control-plane-api/src/lib.rs +++ b/crates/control-plane-api/src/lib.rs @@ -53,7 +53,7 @@ pub use envelope::{Envelope, Locale, MaybeControlClaims}; pub(crate) use server::evaluate_names_authorization; pub use server::{ ApiError, App, AuthZRetry, build_router, - snapshot::{self, Snapshot}, + snapshot::{self, Authorization, Snapshot}, }; // Re-export the GraphQL schema SDL function for flow-client build script diff --git a/crates/control-plane-api/src/server/mod.rs b/crates/control-plane-api/src/server/mod.rs index e3eaef0fe71..406393688d2 100644 --- a/crates/control-plane-api/src/server/mod.rs +++ b/crates/control-plane-api/src/server/mod.rs @@ -39,7 +39,7 @@ pub struct App { pub control_plane_jwt_encode_key: tokens::jwt::EncodingKey, pub pg_pool: sqlx::PgPool, pub publisher: crate::publications::Publisher, - pub snapshot: Arc>, + pub snapshot_watch: Arc>, /// Signing secret for verifying inbound Stripe webhook deliveries. `None` /// when unconfigured, in which case the webhook endpoint fails closed rather /// than trusting any request. See `server::public::stripe_webhooks`. @@ -53,7 +53,7 @@ impl App { jwt_secret: &[u8], pg_pool: sqlx::PgPool, publisher: crate::publications::Publisher, - snapshot: Arc>, + snapshot_watch: Arc>, stripe_webhook_secret: Option, ) -> Self { Self { @@ -63,7 +63,7 @@ impl App { control_plane_jwt_encode_key: tokens::jwt::EncodingKey::from_secret(jwt_secret), pg_pool, publisher, - snapshot, + snapshot_watch, stripe_webhook_secret, } } diff --git a/crates/control-plane-api/src/server/snapshot.rs b/crates/control-plane-api/src/server/snapshot.rs index 9c81400bfc0..2281a60ae3e 100644 --- a/crates/control-plane-api/src/server/snapshot.rs +++ b/crates/control-plane-api/src/server/snapshot.rs @@ -73,6 +73,37 @@ pub struct SnapshotTask { pub data_plane_id: models::Id, } +/// Outcome of an authorization check evaluated against a Snapshot, +/// classified by `Snapshot::resolve_authorization`. +#[must_use] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Authorization { + /// The required grant exists in the Snapshot. + Authorized, + /// The grant is absent and the Snapshot is authoritative for the + /// operation's anchor: the denial is final. + Denied, + /// The grant is absent but the Snapshot predates the anchor: a grant + /// committed before the anchor may not be reflected yet, so the denial is + /// provisional and the operation should retry under a fresher Snapshot. + Stale, +} + +impl Authorization { + /// Collapse to "is authorized?", surfacing a provisional denial as the + /// retryable `AuthorizationSnapshotStale` error which callers + /// (see `validation::is_authz_snapshot_stale`) convert into a retry. + pub fn ok_or_stale(self, catalog_name: &str) -> Result { + match self { + Authorization::Authorized => Ok(true), + Authorization::Denied => Ok(false), + Authorization::Stale => Err(validation::Error::AuthorizationSnapshotStale { + catalog_name: catalog_name.to_string(), + }), + } + } +} + // SnapshotMigration is the state of an underway data-plane migration. #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] pub struct SnapshotMigration { @@ -181,6 +212,68 @@ impl Snapshot { self.taken > (started + Self::TEMPORAL_SKEW) } + /// Classify an already-evaluated authorization check against this + /// Snapshot's freshness: the single three-way policy — authorized / + /// authoritative denial / provisional denial — applied at every snapshot + /// authorization enforcement point. + /// + /// A denial is `Denied` only when this Snapshot was taken after `anchor`, + /// the instant the asking operation started: any grant committed before + /// the anchor is then necessarily reflected. Otherwise it is `Stale` — + /// possibly just unobserved. `None` means the caller has no instant to + /// anchor a staleness claim on, so denials are final. + pub fn resolve_authorization( + &self, + authorized: bool, + anchor: Option, + ) -> Authorization { + if authorized { + Authorization::Authorized + } else if anchor.is_none_or(|anchor| self.taken_after(anchor)) { + Authorization::Denied + } else { + Authorization::Stale + } + } + + /// Evaluate whether `user_id` holds `capability` to `name` under this + /// Snapshot's grants, classified against `anchor` freshness + /// (see `resolve_authorization`). + pub fn user_authorization( + &self, + user_id: uuid::Uuid, + name: &str, + capability: impl Into, + anchor: Option, + ) -> Authorization { + self.resolve_authorization( + tables::UserGrant::is_authorized( + &self.role_grants, + &self.user_grants, + user_id, + name, + capability, + ), + anchor, + ) + } + + /// 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`). + pub fn role_authorization( + &self, + subject: &str, + object: &str, + capability: impl Into, + anchor: Option, + ) -> Authorization { + self.resolve_authorization( + tables::RoleGrant::is_authorized(&self.role_grants, subject, object, capability), + anchor, + ) + } + // Retrieve all tasks whose names start with the given `prefix`. pub fn tasks_by_prefix<'s>( &'s self, @@ -341,9 +434,26 @@ impl Snapshot { }) } + /// Returns the "spec capabilities" of a spec named `catalog_name`: the role + /// grants whose `subject_role` is a prefix of the name — the capabilities the + /// spec holds by virtue of its own name/role. This is only to be used for error + /// reporting to improve error messages. + pub fn spec_capabilities(&self, catalog_name: &str) -> Vec { + self.role_grants + .iter() + .filter(|grant| catalog_name.starts_with(grant.subject_role.as_str())) + .cloned() + .collect() + } + // Minimal interval between Snapshot refreshes. // We will postpone a requested refresh prior to this interval. pub const MIN_REFRESH_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::seconds(20); + /// Re-poll cadence for a queued task which is deferring until its + /// Snapshot is authoritative (see `taken_after`). This equals + /// `MIN_REFRESH_INTERVAL` because that's the soonest a refresh can land: + /// waking sooner burns polls, waking later delays the task. + pub const STALE_RETRY_WAKE: chrono::TimeDelta = Self::MIN_REFRESH_INTERVAL; // Maximum interval between Snapshot refreshes. // We will refresh an older Snapshot in the background. pub const MAX_REFRESH_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::minutes(5); @@ -837,4 +947,187 @@ mod tests { chrono::DateTime::from_timestamp(300_000, 0).unwrap() ); } + + /// `taken_after` is the single definition of "this Snapshot is authoritative + /// for that instant", and every authorization-staleness decision routes + /// through it. The `TEMPORAL_SKEW` allowance and the strictness of the + /// comparison are therefore load-bearing, so pin both. + #[test] + fn test_taken_after_allows_for_temporal_skew() { + let started = chrono::DateTime::from_timestamp(1_000_000, 0).unwrap(); + let at = |offset: chrono::TimeDelta| Snapshot { + taken: started + offset, + ..Snapshot::empty() + }; + + assert!( + !at(chrono::TimeDelta::zero()).taken_after(started), + "a Snapshot taken at the same instant is not authoritative" + ); + assert!( + !at(-Snapshot::TEMPORAL_SKEW).taken_after(started), + "a Snapshot taken before the event is not authoritative" + ); + assert!( + !at(Snapshot::TEMPORAL_SKEW).taken_after(started), + "the skew allowance is exclusive: exactly TEMPORAL_SKEW later is still not authoritative" + ); + assert!( + at(Snapshot::TEMPORAL_SKEW + chrono::TimeDelta::milliseconds(1)).taken_after(started), + "one millisecond past the skew allowance is authoritative" + ); + } + + /// `resolve_authorization` is the shared three-way classifier behind every + /// snapshot authorization enforcement point. Pin its anchor semantics — + /// a denial is authoritative only under a Snapshot postdating the anchor, + /// and a `None` anchor makes denials final — and `ok_or_stale`'s collapse + /// into authorized / dropped / retryable. + #[test] + fn test_resolve_authorization() { + let anchor = chrono::DateTime::from_timestamp(1_000_000, 0).unwrap(); + let stale = Snapshot { + taken: anchor, + ..Snapshot::empty() + }; + let fresh = Snapshot { + taken: anchor + Snapshot::TEMPORAL_SKEW * 2, + ..Snapshot::empty() + }; + + // A held grant is Authorized regardless of freshness. + assert_eq!( + Authorization::Authorized, + stale.resolve_authorization(true, Some(anchor)) + ); + assert_eq!( + Authorization::Authorized, + stale.resolve_authorization(true, None) + ); + + // A denial is authoritative only under a Snapshot postdating the anchor. + assert_eq!( + Authorization::Denied, + fresh.resolve_authorization(false, Some(anchor)) + ); + assert_eq!( + Authorization::Stale, + stale.resolve_authorization(false, Some(anchor)) + ); + + // Without an anchor there is no basis for a staleness claim. + assert_eq!( + Authorization::Denied, + stale.resolve_authorization(false, None) + ); + + assert!(matches!( + Authorization::Authorized.ok_or_stale("acmeCo/task"), + Ok(true) + )); + assert!(matches!( + Authorization::Denied.ok_or_stale("acmeCo/task"), + Ok(false) + )); + assert!(matches!( + Authorization::Stale.ok_or_stale("acmeCo/task"), + Err(validation::Error::AuthorizationSnapshotStale { catalog_name }) + if catalog_name == "acmeCo/task" + )); + } + + /// `spec_capabilities` replaced a SQL-computed `spec_capabilities` column and + /// now renders the "Available grants are:" list in publication authorization + /// errors. It answers "what may a spec named X do, by virtue of its own + /// name?", which is a prefix match on `subject_role` — not on `object_role`, + /// and not scoped to any user. + #[test] + fn test_spec_capabilities() { + let snapshot = Snapshot::build_fixture(None); + let subjects = |name: &str| { + snapshot + .spec_capabilities(name) + .into_iter() + .map(|g| { + ( + g.subject_role.to_string(), + g.object_role.to_string(), + g.capability, + ) + }) + .collect::>() + }; + + // A name under a granted prefix picks up every grant whose subject_role + // is a prefix of it — here both the tenant-wide grants and the more + // specific `bobCo/tires/` one. + insta::assert_debug_snapshot!(subjects("bobCo/tires/source-tread"), @r#" + [ + ( + "bobCo/", + "bobCo/", + Write, + ), + ( + "bobCo/", + "ops/dp/public/", + Read, + ), + ( + "bobCo/tires/", + "acmeCo/shared/", + Read, + ), + ] + "#); + + // The narrower `bobCo/tires/` grant must not leak to a sibling prefix: + // subject matching is by prefix of the *name*, not by shared tenancy. + insta::assert_debug_snapshot!(subjects("bobCo/widgets/source-squash"), @r#" + [ + ( + "bobCo/", + "bobCo/", + Write, + ), + ( + "bobCo/", + "ops/dp/public/", + Read, + ), + ] + "#); + + // `subject_role` is matched as a prefix of the name, so the role itself + // qualifies. + assert_eq!( + vec![( + "bobCo/tires/".to_string(), + "acmeCo/shared/".to_string(), + models::Capability::Read + )], + subjects("bobCo/tires/") + .into_iter() + .filter(|(s, _, _)| s == "bobCo/tires/") + .collect::>(), + ); + + // Grants are not matched by their object_role: `acmeCo/shared/` is + // reachable *from* `bobCo/tires/`, but a spec named `acmeCo/shared/x` + // holds only `acmeCo/`'s own grants. + insta::assert_debug_snapshot!(subjects("acmeCo/shared/thing"), @r#" + [ + ( + "acmeCo/", + "acmeCo/", + Write, + ), + ] + "#); + + assert!( + subjects("unknownCo/thing").is_empty(), + "a name under no granted prefix holds nothing" + ); + } } diff --git a/crates/validation/src/errors.rs b/crates/validation/src/errors.rs index 63fa9b2198f..187a5d17df6 100644 --- a/crates/validation/src/errors.rs +++ b/crates/validation/src/errors.rs @@ -299,6 +299,10 @@ pub enum Error { build_id: models::Id, larger_id: models::Id, }, + #[error( + "authorization for {catalog_name} was evaluated against a control-plane snapshot that is not authoritative for this operation; please retry the operation" + )] + AuthorizationSnapshotStale { catalog_name: String }, #[error( "This spec was updated while you were editing — please refresh and re-apply your changes.\nThis may have been an automated system update. (expected publication ID {expect_id}, actual {actual_id})" )] @@ -413,3 +417,15 @@ impl Error { errors.insert_row(scope.flatten(), anyhow::anyhow!(self)); } } + +/// Returns true if `err` is (or wraps) an [`Error::AuthorizationSnapshotStale`]. +/// This classifies a *retryable* authorization failure: the decision was made +/// against a control-plane snapshot that is not authoritative for the operation, +/// so it should be retried against a fresher snapshot rather than surfaced as a +/// terminal error. +pub fn is_authz_snapshot_stale(err: &anyhow::Error) -> bool { + matches!( + err.downcast_ref::(), + Some(Error::AuthorizationSnapshotStale { .. }) + ) +} diff --git a/crates/validation/src/lib.rs b/crates/validation/src/lib.rs index 1f9c0c31743..ea0b90f0f0c 100644 --- a/crates/validation/src/lib.rs +++ b/crates/validation/src/lib.rs @@ -17,7 +17,7 @@ mod schema; mod storage_mapping; mod test_step; -pub use errors::Error; +pub use errors::{Error, is_authz_snapshot_stale}; pub use noop::NoOpConnectors; /// Portion of the binding namespace reserved for runtime-internal bindings