Authz refactor w/ orthogonal capabilities

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

@@ -66,7 +66,7 @@ impl App {
 pub fn evaluate_names_authorization<'r, Iter, S>(
     snapshot: &Snapshot,
     claims: &crate::ControlClaims,
-    min_capability: models::Capability,
+    min_capability: impl Into<models::AnyCapability>,
     prefixes_or_names: Iter,
 ) -> AuthZResult<()>
 where
@@ -79,14 +79,15 @@ where
         ..
     } = claims;
     let user_email = user_email.as_ref().map(String::as_str).unwrap_or("user");
+    let min_capability: models::AnyCapability = min_capability.into();
 
     for prefix_or_name in prefixes_or_names.into_iter() {
         if !tables::UserGrant::is_authorized(
             &snapshot.role_grants,
             &snapshot.user_grants,
             *user_id,
             prefix_or_name.as_ref(),
-            min_capability,
+            min_capability.clone(),
         ) {
             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}",

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

@@ -57,13 +57,15 @@ mod tests {
                 user_id: *id,
                 object_role: models::Prefix::new(*obj),
                 capability: *cap,
+                capabilities: vec![],
             }
         }));
         let rg = tables::RoleGrants::from_iter(role_grants.iter().map(|(sub, obj, cap)| {
             tables::RoleGrant {
                 subject_role: models::Prefix::new(*sub),
                 object_role: models::Prefix::new(*obj),
                 capability: *cap,
+                capabilities: vec![],
             }
         }));
         (ug, rg)

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

@@ -7,6 +7,10 @@ pub struct PrefixRef {
     pub prefix: models::Prefix,
     /// The capability granted to the user for this prefix.
     pub user_capability: models::Capability,
+    /// Orthogonal capabilities the user has at this prefix, independent of
+    /// the read/write/admin hierarchy. Empty when no orthogonal capabilities
+    /// are granted.
+    pub capabilities: Vec<models::OrthogonalCapability>,
 }
 
 #[derive(Debug, Clone, async_graphql::InputObject)]
@@ -40,16 +44,32 @@ impl PrefixesQuery {
         let env = ctx.data::<crate::Envelope>()?;
 
         connection::query(after, None, first, None, |after, _, first, _| async move {
+            let snapshot = env.snapshot();
+            let user_id = env.claims()?.sub;
+
             let mut all_roles: Vec<PrefixRef> = tables::UserGrant::transitive_roles(
-                &env.snapshot().role_grants,
-                &env.snapshot().user_grants,
-                env.claims()?.sub,
+                &snapshot.role_grants,
+                &snapshot.user_grants,
+                user_id,
             )
             .filter(|grant| grant.capability >= by.min_capability)
             .filter(|grant| after.as_deref().is_none_or(|min| grant.object_role > min))
-            .map(|grant| PrefixRef {
-                prefix: models::Prefix::new(grant.object_role),
-                user_capability: grant.capability,
+            .map(|grant| {
+                let mut capabilities: Vec<models::OrthogonalCapability> = snapshot
+                    .effective_capabilities(
+                        &snapshot.role_grants,
+                        &snapshot.user_grants,
+                        user_id,
+                        grant.object_role,
+                    )
+                    .into_iter()
+                    .collect();
+                capabilities.sort();
+                PrefixRef {
+                    prefix: models::Prefix::new(grant.object_role),
+                    user_capability: grant.capability,
+                    capabilities,
+                }
             })
             .collect();
 
@@ -110,6 +130,7 @@ mod tests {
                                 node {
                                     prefix
                                     userCapability
+                                    capabilities
                                 }
                             }
                         }
@@ -128,18 +149,21 @@ mod tests {
               "edges": [
                 {
                   "node": {
+                    "capabilities": [],
                     "prefix": "aliceCo/",
                     "userCapability": "admin"
                   }
                 },
                 {
                   "node": {
+                    "capabilities": [],
                     "prefix": "aliceCo/data/",
                     "userCapability": "write"
                   }
                 },
                 {
                   "node": {
+                    "capabilities": [],
                     "prefix": "ops/dp/public/",
                     "userCapability": "read"
                   }

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

@@ -1,5 +1,5 @@
 use anyhow::Context;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 // SnapshotData encapsulates all data required to construct a Snapshot.
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
@@ -341,6 +341,28 @@ impl Snapshot {
             })
     }
 
+    pub fn effective_capabilities(
+        &self,
+        role_grants: &[tables::RoleGrant],
+        user_grants: &[tables::UserGrant],
+        user_id: uuid::Uuid,
+        prefix: &str,
+    ) -> HashSet<models::OrthogonalCapability> {
+        let mut bfs_nodes = 0usize;
+        let result = tables::UserGrant::reachable_nodes(role_grants, user_grants, user_id)
+            .inspect(|_| bfs_nodes += 1)
+            .filter(|node| prefix.starts_with(node.object_role))
+            .flat_map(|node| node.capabilities)
+            .collect();
+
+        // bfs_nodes tracks how many grant-graph nodes the BFS expanded.
+        // If a tenant has deep cross-tenant role grants, this number can
+        // grow combinatorially — filter logs on "orthogonal capability BFS"
+        // and watch for spikes to know when to optimize the traversal.
+        tracing::debug!(%user_id, %prefix, bfs_nodes, "orthogonal capability BFS");
+        result
+    }
+
     // 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);
@@ -480,7 +502,8 @@ pub async fn try_fetch(
         SELECT
             g.subject_role AS "subject_role: models::Prefix",
             g.object_role AS "object_role: models::Prefix",
-            g.capability AS "capability: models::Capability"
+            g.capability AS "capability: models::Capability",
+            g.capabilities AS "capabilities: Vec<models::OrthogonalCapability>"
         FROM role_grants g
         "#,
     )
@@ -494,13 +517,14 @@ pub async fn try_fetch(
         SELECT
             g.user_id AS "user_id: uuid::Uuid",
             g.object_role AS "object_role: models::Prefix",
-            g.capability AS "capability: models::Capability"
+            g.capability AS "capability: models::Capability",
+            g.capabilities AS "capabilities: Vec<models::OrthogonalCapability>"
         FROM user_grants g
         "#,
     )
     .fetch_all(pg_pool)
     .await
-    .context("failed to fetch role_grants")?;
+    .context("failed to fetch user_grants")?;
 
     let tasks = sqlx::query_as!(
         SnapshotTask,

crates/control-plane-api/src/server/snapshot_fixture.json

@@ -113,49 +113,58 @@
     {
       "subject_role": "acmeCo/",
       "object_role": "acmeCo/",
-      "capability": "write"
+      "capability": "write",
+      "capabilities": []
     },
     {
       "subject_role": "bobCo/",
       "object_role": "bobCo/",
-      "capability": "write"
+      "capability": "write",
+      "capabilities": []
     },
     {
       "subject_role": "bobCo/tires/",
       "object_role": "acmeCo/shared/",
-      "capability": "read"
+      "capability": "read",
+      "capabilities": []
     },
     {
       "subject_role": "bobCo/",
       "object_role": "ops/dp/public/",
-      "capability": "read"
+      "capability": "read",
+      "capabilities": []
     },
     {
       "subject_role": "aliceCo/",
       "object_role": "ops/dp/public/",
-      "capability": "read"
+      "capability": "read",
+      "capabilities": []
     }
   ],
   "user_grants": [
     {
       "user_id": "20202020-2020-2020-2020-202020202020",
       "object_role": "bobCo/",
-      "capability": "write"
+      "capability": "write",
+      "capabilities": []
     },
     {
       "user_id": "20202020-2020-2020-2020-202020202020",
       "object_role": "bobCo/tires/",
-      "capability": "admin"
+      "capability": "admin",
+      "capabilities": []
     },
     {
       "user_id": "40404040-4040-4040-4040-404040404040",
       "object_role": "aliceCo/",
-      "capability": "admin"
+      "capability": "admin",
+      "capabilities": []
     },
     {
       "user_id": "40404040-4040-4040-4040-404040404040",
       "object_role": "estuary_support/",
-      "capability": "admin"
+      "capability": "admin",
+      "capabilities": []
     }
   ],
   "tasks": [