tables: capability-mask-aware user grant walk

crates/agent/src/discovers.rs

@@ -180,6 +180,7 @@ impl<C: DiscoverConnectors> DiscoverExecutor<C> {
             row.user_id,
             &row.capture_name,
             models::authz::Capability::SpecEdit,
+            models::authz::CapabilityMask::ALL_CAPABILITIES,
         ) {
             // Request an early background refresh: the grant may have been
             // committed after this Snapshot was taken, and cancelling narrows
@@ -209,6 +210,7 @@ impl<C: DiscoverConnectors> DiscoverExecutor<C> {
             row.user_id,
             &row.data_plane_name,
             models::Capability::Read,
+            models::authz::CapabilityMask::ALL_CAPABILITIES,
         )
         .then(|| snapshot.data_plane_by_catalog_name(&row.data_plane_name))
         .flatten()

crates/control-plane-api/src/live_specs/mod.rs

@@ -28,6 +28,7 @@ fn partition_by_authorization<'n>(
                 user_id,
                 name,
                 capability,
+                models::authz::CapabilityMask::ALL_CAPABILITIES,
             )
         });
 

crates/control-plane-api/src/server/authorize_user_collection.rs

@@ -71,6 +71,7 @@ fn evaluate_authorization(
         *user_id,
         collection_name,
         capability,
+        models::authz::CapabilityMask::ALL_CAPABILITIES,
     ) {
         return Err(tonic::Status::permission_denied(format!(
             "{user_email} is not authorized to {collection_name} for {capability:?}",
@@ -85,6 +86,7 @@ fn evaluate_authorization(
             *user_id,
             "estuary_support/",
             models::Capability::Admin,
+            models::authz::CapabilityMask::ALL_CAPABILITIES,
         );
 
         if !has_support_access {

crates/control-plane-api/src/server/authorize_user_prefix.rs

@@ -88,6 +88,7 @@ fn evaluate_authorization(
         *user_id,
         prefix,
         capability,
+        models::authz::CapabilityMask::ALL_CAPABILITIES,
     ) {
         return Err(tonic::Status::permission_denied(format!(
             "{user_email} is not authorized to {prefix} for {capability:?}",
@@ -102,6 +103,7 @@ fn evaluate_authorization(
             *user_id,
             "estuary_support/",
             models::Capability::Admin,
+            models::authz::CapabilityMask::ALL_CAPABILITIES,
         );
 
         if !has_support_access {
@@ -117,6 +119,7 @@ fn evaluate_authorization(
         *user_id,
         data_plane_name,
         models::Capability::Read,
+        models::authz::CapabilityMask::ALL_CAPABILITIES,
     ) {
         return Err(tonic::Status::permission_denied(format!(
             "{user_email} is not authorized to {data_plane_name}",

crates/control-plane-api/src/server/authorize_user_task.rs

@@ -95,6 +95,7 @@ fn evaluate_authorization(
         *user_id,
         task_name,
         capability,
+        models::authz::CapabilityMask::ALL_CAPABILITIES,
     ) {
         return Err(tonic::Status::permission_denied(format!(
             "{user_email} is not authorized to {task_name} for {capability:?}",
@@ -109,6 +110,7 @@ fn evaluate_authorization(
             *user_id,
             "estuary_support/",
             models::Capability::Admin,
+            models::authz::CapabilityMask::ALL_CAPABILITIES,
         );
 
         if !has_support_access {

crates/control-plane-api/src/server/mod.rs

@@ -125,6 +125,7 @@ where
             *user_id,
             prefix_or_name.as_ref(),
             min_capability,
+            models::authz::CapabilityMask::ALL_CAPABILITIES,
         ) {
             return Err(tonic::Status::permission_denied(format!(
                 "{user_email} is not authorized to access prefix or name '{prefix_or_name}' with required capability {min_capability}",
@@ -155,6 +156,7 @@ where
                 &snapshot.user_grants,
                 claims.sub,
                 &prefix,
+                models::authz::CapabilityMask::ALL_CAPABILITIES,
             );
             attach(prefix, capability)
         })

crates/control-plane-api/src/server/public/graphql/authorized_prefixes.rs

@@ -19,13 +19,18 @@ pub(super) fn authorized_prefixes(
 
     // BTreeMap iteration from reachable_prefixes is already prefix-sorted,
     // so the parent-prune step below can run directly on it.
-    let prefixes = tables::UserGrant::reachable_prefixes(role_grants, user_grants, user_id)
-        .into_iter()
-        .filter(|(prefix, _)| {
-            prefix_filter.is_none_or(|pf| prefix.starts_with(pf) || pf.starts_with(*prefix))
-        })
-        .filter(|(_, (bits, _))| bits.is_superset(min_bits))
-        .map(|(prefix, _)| prefix.to_string());
+    let prefixes = tables::UserGrant::reachable_prefixes(
+        role_grants,
+        user_grants,
+        user_id,
+        models::authz::CapabilityMask::ALL_CAPABILITIES,
+    )
+    .into_iter()
+    .filter(|(prefix, _)| {
+        prefix_filter.is_none_or(|pf| prefix.starts_with(pf) || pf.starts_with(*prefix))
+    })
+    .filter(|(_, (bits, _))| bits.is_superset(min_bits))
+    .map(|(prefix, _)| prefix.to_string());
 
     let mut pruned: Vec<String> = Vec::new();
     for p in prefixes {
@@ -237,7 +242,12 @@ mod tests {
         ]);
         let rg = tables::RoleGrants::new();
 
-        let reachable = tables::UserGrant::reachable_prefixes(&rg, &ug, ALICE);
+        let reachable = tables::UserGrant::reachable_prefixes(
+            &rg,
+            &ug,
+            ALICE,
+            models::authz::CapabilityMask::ALL_CAPABILITIES,
+        );
         assert_eq!(
             reachable["acmeCo/"].0,
             CapabilityBundle::Editor.capabilities() | CapabilityBundle::TeamAdmin.capabilities(),
@@ -274,7 +284,12 @@ mod tests {
             },
         ]);
 
-        let reachable = tables::UserGrant::reachable_prefixes(&rg, &ug, ALICE);
+        let reachable = tables::UserGrant::reachable_prefixes(
+            &rg,
+            &ug,
+            ALICE,
+            models::authz::CapabilityMask::ALL_CAPABILITIES,
+        );
         assert_eq!(
             reachable["sharedCo/"].0,
             CapabilityBundle::Editor.capabilities() | CapabilityBundle::TeamAdmin.capabilities(),

crates/control-plane-api/src/server/public/graphql/data_planes.rs

@@ -518,6 +518,7 @@ impl DataPlanesQuery {
                         claims.sub,
                         &dp.data_plane_name,
                         models::Capability::Read,
+                        models::authz::CapabilityMask::ALL_CAPABILITIES,
                     )
             })
             .collect();

crates/control-plane-api/src/server/public/graphql/mod.rs

@@ -70,6 +70,7 @@ fn may_access(
         env.claims()?.sub,
         name,
         capability,
+        models::authz::CapabilityMask::ALL_CAPABILITIES,
     ))
 }
 

crates/control-plane-api/src/server/public/graphql/prefixes.rs

@@ -59,6 +59,7 @@ impl PrefixesQuery {
                 &snapshot.role_grants,
                 &snapshot.user_grants,
                 user_id,
+                models::authz::CapabilityMask::ALL_CAPABILITIES,
             );
             // Cursor pagination: BTreeMap::range jumps directly to the
             // first key strictly greater than the previous page's last

crates/control-plane-api/src/server/public/graphql/storage_mappings.rs

@@ -537,6 +537,7 @@ fn check_authorization(
         *user_id,
         catalog_prefix,
         models::Capability::Admin,
+        models::authz::CapabilityMask::ALL_CAPABILITIES,
     ) {
         return Err(tonic::Status::permission_denied(format!(
             "{user_email} is not an authorized as an Admin of catalog prefix '{catalog_prefix}'",
@@ -779,6 +780,7 @@ impl StorageMappingsQuery {
                     &snapshot.user_grants,
                     claims.sub,
                     &row.catalog_prefix,
+                    models::authz::CapabilityMask::ALL_CAPABILITIES,
                 )
                 .ok_or_else(|| {
                     async_graphql::Error::new(format!(

crates/control-plane-api/src/server/public/status.rs

@@ -53,6 +53,7 @@ pub(crate) async fn handle_get_status(
                     claims.sub,
                     name,
                     models::Capability::Read,
+                    models::authz::CapabilityMask::ALL_CAPABILITIES,
                 )
             })
             .collect::<Vec<_>>();

crates/models/src/authorizations.rs

@@ -330,7 +330,7 @@ mod test {
         // variant is added.
         for (claim, mask) in &outcomes {
             if claim.is_none() {
-                assert_eq!(*mask, CapabilityMask::UNMASKED);
+                assert_eq!(*mask, CapabilityMask::ALL_CAPABILITIES);
             }
         }
         let bounded: Vec<_> = outcomes
@@ -485,9 +485,9 @@ mod test {
 
         // A populated mask serializes its names verbatim — including names
         // this binary doesn't recognize — and they survive a round trip
-        // intact. Carry-through is load-bearing: an upgrade token's
-        // unrecognized names must re-mint unchanged rather than being
-        // dropped by whichever instance happens to re-sign it.
+        // intact. Carry-through is load-bearing in a mixed-version fleet:
+        // names minted by a newer instance must pass through an older one
+        // unchanged rather than being silently dropped.
         let masked = ControlClaims {
             capability_mask: Some(vec!["SpecEdit".to_string(), "FutureCapability".to_string()]),
             ..masked

crates/models/src/authz.rs

@@ -92,7 +92,7 @@ impl Capability {
 /// The mask is an enable/disable filter, never a grant: `apply` is pure
 /// intersection, so naming a capability the user doesn't hold conveys
 /// nothing, while omitting one they do hold disables it. An unmasked bearer
-/// simply carries the full set ([`Self::UNMASKED`]) and intersects as the
+/// simply carries the full set ([`Self::ALL_CAPABILITIES`]) and intersects as the
 /// identity.
 ///
 /// This is a newtype over [`CapabilitySet`] rather than a bare set because
@@ -103,8 +103,9 @@ impl Capability {
 /// This type answers *what may be exercised*, never *whether the bearer is
 /// masked*. A token whose mask happens to enable everything is still a
 /// deliberately-reduced credential, and surfaces that fail closed for masked
-/// bearers (the `/admin` endpoints, the mint) must key on the claim's
-/// presence — `capability_mask.is_some()` — and never on [`Self::is_all`].
+/// bearers (such as the `/admin` endpoints) must key on the claim's
+/// presence — `capability_mask.is_some()` — and never on
+/// [`Self::has_all_capabilities`].
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct CapabilityMask(CapabilitySet);
 
@@ -116,8 +117,8 @@ impl CapabilityMask {
     ///
     /// There is deliberately no `Default` and no `From<CapabilitySet>`:
     /// every caller must name its mask, and constructing an unrestricted
-    /// one must be a visible, greppable choice.
-    pub const UNMASKED: Self = Self(CapabilitySet::all());
+    /// one must be a visible choice.
+    pub const ALL_CAPABILITIES: Self = Self(CapabilitySet::all());
 
     /// A mask enabling exactly `set`. An empty set is valid and yields a
     /// token which authenticates an identity but authorizes nothing.
@@ -127,16 +128,16 @@ impl CapabilityMask {
 
     /// Build a mask from a token's verified `capability_mask` claim.
     ///
-    /// An absent claim is [`Self::UNMASKED`]; a present claim enables the
-    /// union of the capability bits of its recognized [`CapabilityBundle`]
-    /// names, and that includes an empty list — "no mask" and "an empty
-    /// mask" are distinct on the wire and the difference is load-bearing.
-    /// Unrecognized names contribute nothing, so a claim naming only names
-    /// we don't know bounds the token to nothing at all; see
-    /// [`CapabilityBundle::from_name`].
+    /// An absent claim is [`Self::ALL_CAPABILITIES`]; a present claim
+    /// enables the union of the capability bits of its recognized
+    /// [`CapabilityBundle`] names, and that includes an empty list — "no
+    /// mask" and "an empty mask" are distinct on the wire and the
+    /// difference is load-bearing. Unrecognized names contribute nothing,
+    /// so a claim naming only names we don't know bounds the token to
+    /// nothing at all; see [`CapabilityBundle::from_name`].
     pub fn from_claim(mask: Option<&[String]>) -> Self {
         let Some(mask) = mask else {
-            return Self::UNMASKED;
+            return Self::ALL_CAPABILITIES;
         };
         Self(
             mask.iter()
@@ -149,10 +150,10 @@ impl CapabilityMask {
     /// Attenuate `capabilities` to this mask.
     ///
     /// Apply this at each node emission of the user grant walk, never to the
-    /// walk's result: the mask has to gate `Delegate` itself, so that a mask
-    /// without it confines the token to direct user grants, and it must not
-    /// be re-widened by `Assume`, which makes all of an edge's bits
-    /// delegatable as it passes through.
+    /// walk's result: the mask has to gate traversal itself, so that a mask
+    /// without `Delegate` (and `Assume`) confines the token to direct user
+    /// grants, and it must not be re-widened by `Assume`, which makes all of
+    /// an edge's bits delegatable as it passes through.
     pub fn apply(self, capabilities: CapabilitySet) -> CapabilitySet {
         capabilities & self.0
     }
@@ -164,7 +165,7 @@ impl CapabilityMask {
     /// which is safe when the mask hides nothing. It must NEVER stand in
     /// for "is this bearer unmasked?": that is a property of the claim
     /// (`capability_mask.is_some()`), not of this value.
-    pub fn is_all(self) -> bool {
+    pub fn has_all_capabilities(self) -> bool {
         self.0 == CapabilitySet::all()
     }
 }
@@ -477,7 +478,10 @@ mod test {
     #[test]
     fn test_capability_mask_from_claim() {
         // An absent claim is an unmasked token: the full set.
-        assert_eq!(CapabilityMask::from_claim(None), CapabilityMask::UNMASKED);
+        assert_eq!(
+            CapabilityMask::from_claim(None),
+            CapabilityMask::ALL_CAPABILITIES
+        );
 
         let cases = [
             // Single-capability bundle names enable exactly the bit they
@@ -518,7 +522,7 @@ mod test {
         );
         assert_ne!(
             CapabilityMask::from_claim(Some(&[])),
-            CapabilityMask::UNMASKED,
+            CapabilityMask::ALL_CAPABILITIES,
         );
 
         insta::assert_debug_snapshot!(masks, @r"
@@ -554,10 +558,10 @@ mod test {
 
         // The unmasked mask is the identity, and enabling every capability
         // is the same thing by construction.
-        assert_eq!(CapabilityMask::UNMASKED.apply(editor), editor);
+        assert_eq!(CapabilityMask::ALL_CAPABILITIES.apply(editor), editor);
         assert_eq!(
             CapabilityMask::bounded(CapabilitySet::all()),
-            CapabilityMask::UNMASKED,
+            CapabilityMask::ALL_CAPABILITIES,
         );
 
         // A mask intersects: it can only ever disable bits, and bits it
@@ -571,8 +575,8 @@ mod test {
             CapabilityMask::bounded(CapabilitySet::empty()).apply(editor),
             CapabilitySet::empty(),
         );
-        assert!(!CapabilityMask::bounded(editor).is_all());
-        assert!(CapabilityMask::UNMASKED.is_all());
+        assert!(!CapabilityMask::bounded(editor).has_all_capabilities());
+        assert!(CapabilityMask::ALL_CAPABILITIES.has_all_capabilities());
     }
 
     // GraphQL's `CapabilityBit` vocabulary must stay a subset of the claim