authz: scope a request's authority to one branch of the grant graph

crates/agent/src/discovers.rs

@@ -177,7 +177,7 @@ impl<C: DiscoverConnectors> DiscoverExecutor<C> {
         if !tables::UserGrant::is_authorized(
             &snapshot.role_grants,
             &snapshot.user_grants,
-            row.user_id,
+            tables::Principal::unscoped(row.user_id),
             &row.capture_name,
             models::authz::Capability::SpecEdit,
         ) {
@@ -206,7 +206,7 @@ impl<C: DiscoverConnectors> DiscoverExecutor<C> {
         let Some(data_plane) = tables::UserGrant::is_authorized(
             &snapshot.role_grants,
             &snapshot.user_grants,
-            row.user_id,
+            tables::Principal::unscoped(row.user_id),
             &row.data_plane_name,
             models::Capability::Read,
         )
@@ -313,8 +313,10 @@ async fn prepare_discover<'a>(
     // conveying SpecEdit convey CatalogRead too. Under a SpecEdit-only
     // grant the discover proceeds with the live capture treated as absent.
     let name = &[capture_name.to_string()];
+    // Discovers authorize against the user recorded on the job, which is not
+    // tied to the request that enqueued it, so there is no scope prefix here.
     let live = live_specs::get_live_specs_filtered(
-        user_id,
+        tables::Principal::unscoped(user_id),
         name,
         models::authz::Capability::CatalogRead,
         snapshot,

crates/agent/src/integration_tests/harness.rs

@@ -1696,6 +1696,7 @@ impl TestHarness {
             pg_pool: self.pool.clone(),
             refresh: app.snapshot.token(),
             retry_after: tokens::DateTime::UNIX_EPOCH,
+            scope_prefix: None,
             started: tokens::now(),
             locale: control_plane_api::Locale::EnUS,
         };

crates/control-plane-api/README.md

@@ -49,3 +49,33 @@ Tests use `insta` for snapshot testing. To automatically accept updated snapshot
 ```bash
 INSTA_UPDATE=always cargo test -p control-plane-api -- --test-threads=1
 ```
+
+## Scoping a request to one branch of the grant graph
+
+Requests may narrow their own authority with the `X-Estuary-Scope-Prefix`
+header, naming a catalog prefix:
+
+```
+X-Estuary-Scope-Prefix: acmeCo/
+```
+
+Authorization then considers only prefixes reachable both from the user's
+grants and from that prefix, including prefixes it reaches through
+`role_grants`. So a user who admins `acmeCo/` and `betaCo/`, where `acmeCo/`
+holds a role grant to `charlieCo/`, sees `acmeCo/` and `charlieCo/` under this
+header and does not see `betaCo/`.
+
+Because the result is an intersection with the user's own grants, the header
+can only remove authority. A prefix the user cannot reach yields nothing
+rather than access to it. That is why a client may set the header freely — the
+dashboard uses it to let a user pick which tenant they are working in, and to
+switch without re-authenticating.
+
+Omit the header to request the user's full authority. An empty or malformed
+value is rejected with `invalid_argument` rather than silently matching
+everything or nothing.
+
+Handlers reach this through `Envelope::principal`, which pairs the
+authenticated user with the scope. It is the only input the
+`tables::UserGrant` authorization functions accept, so a handler cannot honor
+the token while overlooking the scope.

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

@@ -316,8 +316,11 @@ impl<C: DiscoverConnectors> DiscoverHandler<C> {
             .collect::<Vec<_>>();
 
         let live = if filter_user_authz {
+            // Discovers authorize against the user recorded on the job, which
+            // is not tied to the request that enqueued it, so there is no
+            // scope prefix to narrow by here.
             crate::live_specs::get_live_specs_filtered(
-                user_id,
+                tables::Principal::unscoped(user_id),
                 &collection_names,
                 models::authz::Capability::CatalogRead,
                 snapshot,

crates/control-plane-api/src/envelope.rs

@@ -48,6 +48,15 @@ pub struct Envelope {
     pub original_uri: axum::http::Uri,
     /// The verified control-plane claims, if any.
     pub maybe_claims: MaybeControlClaims,
+    /// Catalog prefix which narrows this request's authority, taken from the
+    /// `X-Estuary-Scope-Prefix` header.
+    ///
+    /// Authorization considers only prefixes reachable both from the user's
+    /// grants and from this prefix, so the header can only ever remove
+    /// authority. That makes it safe for a client to set freely, which is how
+    /// the web UI lets a user choose which tenant they are working in and
+    /// switch between them without re-authenticating.
+    pub scope_prefix: Option<models::Prefix>,
     /// If provided, the `retryAfter` query parameter attached to the request.
     /// This parameter is used to detect clients that don't honor the Retry-After
     /// header sent with 307 Temporary Redirect responses.
@@ -70,12 +79,73 @@ pub struct Envelope {
     pub locale: Locale,
 }
 
+/// Header naming the catalog prefix which narrows a request's authority.
+pub const SCOPE_PREFIX_HEADER: &str = "x-estuary-scope-prefix";
+
+/// Reads and validates the scope prefix header.
+///
+/// A malformed value is rejected rather than passed through, because an
+/// unparseable prefix would match no grant and present as an authorization
+/// failure that gives the client no way to tell a typo from a missing grant.
+/// An empty value is rejected for the same reason in the other direction: it
+/// would match every prefix and silently do nothing.
+fn parse_scope_prefix(
+    headers: &axum::http::HeaderMap,
+) -> tonic::Result<Option<models::Prefix>, tonic::Status> {
+    use validator::Validate;
+
+    let Some(value) = headers.get(SCOPE_PREFIX_HEADER) else {
+        return Ok(None);
+    };
+    let value = value.to_str().map_err(|_err| {
+        tonic::Status::invalid_argument(format!("{SCOPE_PREFIX_HEADER} header is not valid UTF-8"))
+    })?;
+
+    if value.is_empty() {
+        return Err(tonic::Status::invalid_argument(format!(
+            "{SCOPE_PREFIX_HEADER} header is empty; omit it to request unscoped authority"
+        )));
+    }
+
+    let prefix = models::Prefix::new(value);
+    if let Err(err) = prefix.validate() {
+        return Err(tonic::Status::invalid_argument(format!(
+            "{SCOPE_PREFIX_HEADER} header '{value}' is not a valid catalog prefix: {err}"
+        )));
+    }
+    Ok(Some(prefix))
+}
+
 impl Envelope {
     /// Returns verified ControlClaims or an unauthenticated error.
     pub fn claims(&self) -> tonic::Result<&crate::ControlClaims> {
         self.maybe_claims.result()
     }
 
+    /// Returns the principal on whose behalf this request is authorized:
+    /// the authenticated user, narrowed by `scope_prefix` if it was given.
+    ///
+    /// Every user authorization check goes through a `Principal`, so a handler
+    /// cannot honor the token while overlooking the scope.
+    pub fn principal(&self) -> tonic::Result<tables::Principal<'_>> {
+        let claims = self.claims()?;
+
+        Ok(tables::Principal {
+            user_id: claims.sub,
+            scope: self.scope_prefix.as_ref().map(|p| p.as_str()),
+        })
+    }
+
+    /// The authenticated user's email, for inclusion in error messages.
+    /// Falls back to "user" when the token carries no email.
+    pub fn user_email(&self) -> &str {
+        self.maybe_claims
+            .result()
+            .ok()
+            .and_then(|claims| claims.email.as_deref())
+            .unwrap_or("user")
+    }
+
     /// Returns the request's associated Snapshot.
     pub fn snapshot(&self) -> &crate::Snapshot {
         self.refresh.result().expect("Snapshot refresh never fails")
@@ -263,6 +333,8 @@ impl axum::extract::FromRequestParts<Arc<crate::App>> for Envelope {
                 None => MaybeControlClaims::with_unauthenticated(),
             };
 
+            let scope_prefix = parse_scope_prefix(&parts.headers)?;
+
             // Placeholder. In the future, we should determine this value from
             // the request headers (e.g. Accept-Language and/or the auth token).
             // For now, we hard code it, because we don't have translations for
@@ -271,6 +343,7 @@ impl axum::extract::FromRequestParts<Arc<crate::App>> for Envelope {
 
             Ok(Envelope {
                 maybe_claims,
+                scope_prefix,
                 retry_after: retry_after.unwrap_or(tokens::DateTime::UNIX_EPOCH),
                 refresh: state.snapshot.token(),
                 started: started.unwrap_or_else(|| tokens::now()),
@@ -285,3 +358,51 @@ impl axum::extract::FromRequestParts<Arc<crate::App>> for Envelope {
 // Empty impl allows aide to generate OpenAPI specs for handlers using this extractor.
 // The extractor is an internal detail and doesn't appear in the API documentation.
 impl aide::operation::OperationInput for Envelope {}
+
+#[cfg(test)]
+mod test {
+    use super::{SCOPE_PREFIX_HEADER, parse_scope_prefix};
+
+    fn parse(value: &str) -> tonic::Result<Option<String>, tonic::Status> {
+        let mut headers = axum::http::HeaderMap::new();
+        headers.insert(
+            axum::http::HeaderName::from_static(SCOPE_PREFIX_HEADER),
+            axum::http::HeaderValue::from_str(value).expect("test header value"),
+        );
+        parse_scope_prefix(&headers).map(|p| p.map(|p| p.to_string()))
+    }
+
+    #[test]
+    fn test_absent_header_is_unscoped() {
+        let headers = axum::http::HeaderMap::new();
+        assert_eq!(parse_scope_prefix(&headers).unwrap(), None);
+    }
+
+    #[test]
+    fn test_valid_prefixes() {
+        for case in ["acmeCo/", "acmeCo/team/", "acmeCo/one/two/th.ree/"] {
+            assert_eq!(parse(case).unwrap().as_deref(), Some(case));
+        }
+    }
+
+    #[test]
+    fn test_rejected_values() {
+        // Empty would scope to everything, which the caller should express by
+        // omitting the header.
+        assert_eq!(
+            parse("").unwrap_err().code(),
+            tonic::Code::InvalidArgument,
+            "empty value must be rejected",
+        );
+
+        // A catalog *name* is not a prefix: prefixes must end in '/'. Passing
+        // one is a client error rather than a scope matching nothing.
+        for case in ["acmeCo", "/acmeCo/", "acmeCo//team/", "acmeCo/sp ace/", "/"] {
+            assert_eq!(
+                parse(case).unwrap_err().code(),
+                tonic::Code::InvalidArgument,
+                "'{case}' must be rejected",
+            );
+        }
+    }
+}

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

@@ -10,12 +10,12 @@ pub use db::{
     fetch_live_spec_names_by_prefix, fetch_live_specs, hard_delete_live_spec,
 };
 
-/// Partitions the requested `names` by whether the user holds `capability`
+/// Partitions the requested `names` by whether `principal` holds `capability`
 /// to them, evaluated against the authorization `snapshot`. Returns
 /// `(authorized, denied)` as references into `names`, each sorted and
 /// deduplicated.
 fn partition_by_authorization<'n>(
-    user_id: Uuid,
+    principal: tables::Principal<'_>,
     names: &'n [String],
     capability: models::authz::CapabilitySet,
     snapshot: &crate::Snapshot,
@@ -25,7 +25,7 @@ fn partition_by_authorization<'n>(
             tables::UserGrant::is_authorized(
                 &snapshot.role_grants,
                 &snapshot.user_grants,
-                user_id,
+                principal,
                 name,
                 capability,
             )
@@ -47,19 +47,20 @@ fn partition_by_authorization<'n>(
 /// The `snapshot` is trusted as-is: a grant committed after it was taken is
 /// invisible until the watch's own background refresh cadence picks it up.
 pub async fn get_live_specs_filtered(
-    user_id: Uuid,
+    principal: tables::Principal<'_>,
     names: &[String],
     capability: impl Into<models::authz::CapabilitySet>,
     snapshot: &crate::Snapshot,
     db: &sqlx::PgPool,
 ) -> anyhow::Result<tables::LiveCatalog> {
     let (authorized, denied) =
-        partition_by_authorization(user_id, names, capability.into(), snapshot);
+        partition_by_authorization(principal, names, capability.into(), snapshot);
 
     if !denied.is_empty() {
+        let user_id = principal.user_id;
         tracing::debug!(?denied, %user_id, "filtered unauthorized specs from fetch");
     }
-    get_live_specs_unfiltered(user_id, &authorized, db).await
+    get_live_specs_unfiltered(principal.user_id, &authorized, db).await
 }
 
 /// Fetches live specs as a `tables::LiveCatalog` without any authorization
@@ -186,7 +187,12 @@ mod test {
             "acmeCo/shared/collection".to_string(),
             "bobCo/tires/capture".to_string(),
         ];
-        let (authorized, denied) = partition_by_authorization(bob, &names, capability, &snapshot);
+        let (authorized, denied) = partition_by_authorization(
+            tables::Principal::unscoped(bob),
+            &names,
+            capability,
+            &snapshot,
+        );
         assert_eq!(authorized, names);
         assert!(denied.is_empty());
 
@@ -199,7 +205,12 @@ mod test {
             "acmeCo/private/collection".to_string(),
             "aliceCo/anvils/pings".to_string(),
         ];
-        let (authorized, denied) = partition_by_authorization(bob, &names, capability, &snapshot);
+        let (authorized, denied) = partition_by_authorization(
+            tables::Principal::unscoped(bob),
+            &names,
+            capability,
+            &snapshot,
+        );
         assert_eq!(authorized, vec!["bobCo/tires/capture"]);
         assert_eq!(
             denied,

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

@@ -18,8 +18,13 @@ pub async fn authorize_user_collection(
             tokens::DateTime::from_timestamp_secs(1 + started_unix as i64).unwrap_or_default();
     }
 
-    let policy_result =
-        evaluate_authorization(env.snapshot(), env.claims()?, &collection, capability);
+    let policy_result = evaluate_authorization(
+        env.snapshot(),
+        env.principal()?,
+        env.user_email(),
+        &collection,
+        capability,
+    );
 
     // Legacy: if `started_unix` was set then use a custom 200 response for client-side retries.
     let (expiry, (encoding_key, mut claims, broker_address, journal_name_prefix)) =
@@ -49,7 +54,8 @@ pub async fn authorize_user_collection(
 
 fn evaluate_authorization(
     snapshot: &crate::Snapshot,
-    claims: &crate::ControlClaims,
+    principal: tables::Principal<'_>,
+    user_email: &str,
     collection_name: &models::Collection,
     capability: models::Capability,
 ) -> crate::AuthZResult<(
@@ -58,17 +64,10 @@ fn evaluate_authorization(
     String,
     String,
 )> {
-    let models::authorizations::ControlClaims {
-        sub: user_id,
-        email: user_email,
-        ..
-    } = claims;
-    let user_email = user_email.as_ref().map(String::as_str).unwrap_or("user");
-
     if !tables::UserGrant::is_authorized(
         &snapshot.role_grants,
         &snapshot.user_grants,
-        *user_id,
+        principal,
         collection_name,
         capability,
     ) {
@@ -82,7 +81,7 @@ fn evaluate_authorization(
         let has_support_access = tables::UserGrant::is_authorized(
             &snapshot.role_grants,
             &snapshot.user_grants,
-            *user_id,
+            principal,
             "estuary_support/",
             models::Capability::Admin,
         );
@@ -119,7 +118,7 @@ fn evaluate_authorization(
         exp: 0, // Filled later.
         iat: 0, // Filled later.
         iss: data_plane.data_plane_fqdn.clone(),
-        sub: user_id.to_string(),
+        sub: principal.user_id.to_string(),
         sel: proto_gazette::broker::LabelSelector {
             include: Some(labels::build_set([
                 ("name:prefix", collection.journal_template_name.as_str()),
@@ -373,7 +372,13 @@ mod tests {
             email,
         };
 
-        match evaluate_authorization(&snapshot, &claims, &collection, capability) {
+        match evaluate_authorization(
+            &snapshot,
+            tables::Principal::unscoped(claims.sub),
+            "test@example.com",
+            &collection,
+            capability,
+        ) {
             Ok((cordon_at, (_key, mut data_claims, broker_address, journal_name_prefix))) => {
                 // Zero out timestamps for stable snapshots.
                 data_claims.iat = 0;