feat: Support Authorizers which cannot list projects (lakekeeper#1481)

Author: c-thiel

crates/authz-openfga/src/authorizer.rs

@@ -290,6 +290,22 @@ impl Authorizer for OpenFGAAuthorizer {
         .map_err(Into::into)
     }
 
+    async fn are_allowed_project_actions_impl(
+        &self,
+        metadata: &RequestMetadata,
+        projects_with_actions: &[(&ProjectId, Self::ProjectAction)],
+    ) -> std::result::Result<Vec<bool>, AuthorizationBackendUnavailable> {
+        let items: Vec<_> = projects_with_actions
+            .iter()
+            .map(|(project, a)| CheckRequestTupleKey {
+                user: metadata.actor().to_openfga(),
+                relation: a.to_string(),
+                object: project.to_openfga(),
+            })
+            .collect();
+        self.batch_check(items).await.map_err(Into::into)
+    }
+
     async fn is_allowed_warehouse_action_impl(
         &self,
         metadata: &RequestMetadata,

crates/lakekeeper/src/api/management/v1/project.rs

@@ -211,18 +211,49 @@ pub trait Service<C: CatalogStore, A: Authorizer, S: SecretStore> {
     ) -> Result<ListProjectsResponse> {
         // ------------------- AuthZ -------------------
         let authorizer = context.v1_state.authz;
-        let projects = authorizer.list_projects(&request_metadata).await?;
+        let authz_projects_response = authorizer.list_projects(&request_metadata).await?;
 
         // ------------------- Business Logic -------------------
-        let project_id_filter = match projects {
-            AuthZListProjectsResponse::All => None,
+        let authz_list_unsupported =
+            authz_projects_response == AuthZListProjectsResponse::Unsupported;
+        let project_id_filter = match authz_projects_response {
+            AuthZListProjectsResponse::All | AuthZListProjectsResponse::Unsupported => None,
             AuthZListProjectsResponse::Projects(projects) => Some(projects),
         };
         let mut trx = C::Transaction::begin_read(context.v1_state.catalog).await?;
-
         let projects = C::list_projects(project_id_filter, trx.transaction()).await?;
         trx.commit().await?;
 
+        let projects = if authz_list_unsupported {
+            tracing::debug!(
+                "Authorization backend does not support listing projects, filtering per project."
+            );
+            let decisions = authorizer
+                .are_allowed_project_actions_vec(
+                    &request_metadata,
+                    &projects
+                        .iter()
+                        .map(|p| (&p.project_id, CatalogProjectAction::CanGetMetadata))
+                        .collect::<Vec<_>>(),
+                )
+                .await?;
+            projects
+                .into_iter()
+                .zip(decisions.into_inner())
+                .filter_map(
+                    |(project, is_allowed)| {
+                        if is_allowed {
+                            Some(project)
+                        } else {
+                            None
+                        }
+                    },
+                )
+                .collect()
+        } else {
+            projects
+        };
+
         Ok(ListProjectsResponse {
             projects: projects
                 .into_iter()

crates/lakekeeper/src/service/authz/mod.rs

@@ -241,10 +241,14 @@ where
     async fn bootstrap(&self, metadata: &RequestMetadata, is_operator: bool) -> Result<()>;
 
     /// Return Err only for internal errors.
+    /// If unsupported is returned, Lakekeeper will run checks for every project individually using
+    /// `are_allowed_project_actions`.
     async fn list_projects_impl(
         &self,
-        metadata: &RequestMetadata,
-    ) -> std::result::Result<ListProjectsResponse, AuthorizationBackendUnavailable>;
+        _metadata: &RequestMetadata,
+    ) -> std::result::Result<ListProjectsResponse, AuthorizationBackendUnavailable> {
+        Ok(ListProjectsResponse::Unsupported)
+    }
 
     /// Search users
     async fn can_search_users_impl(&self, metadata: &RequestMetadata) -> Result<bool>;
@@ -323,6 +327,29 @@ where
         action: Self::ProjectAction,
     ) -> std::result::Result<bool, AuthorizationBackendUnavailable>;
 
+    async fn are_allowed_project_actions_impl(
+        &self,
+        metadata: &RequestMetadata,
+        projects_with_actions: &[(&ProjectId, Self::ProjectAction)],
+    ) -> std::result::Result<Vec<bool>, AuthorizationBackendUnavailable> {
+        let n_inputs = projects_with_actions.len();
+        let futures: Vec<_> = projects_with_actions
+            .iter()
+            .map(|(project, a)| async move {
+                self.is_allowed_project_action(metadata, project, *a)
+                    .await
+                    .map(MustUse::into_inner)
+            })
+            .collect();
+        let results = try_join_all(futures).await?;
+        debug_assert_eq!(
+            results.len(),
+            n_inputs,
+            "are_allowed_project_actions_impl to return as many results as provided inputs"
+        );
+        Ok(results)
+    }
+
     /// Return Ok(true) if the action is allowed, otherwise return Ok(false).
     /// Return Err for internal errors.
     async fn is_allowed_warehouse_action_impl(

crates/lakekeeper/src/service/authz/project.rs

@@ -18,12 +18,14 @@ where
 
 impl ProjectAction for CatalogProjectAction {}
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ListProjectsResponse {
     /// List of projects that the user is allowed to see.
     Projects(HashSet<ProjectId>),
     /// The user is allowed to see all projects.
     All,
+    /// Unsupported by the authorization backend.
+    Unsupported,
 }
 
 // --------------------------- Errors ---------------------------
@@ -96,6 +98,37 @@ pub trait AuthZProjectOps: Authorizer {
         }
     }
 
+    async fn are_allowed_project_actions_vec<A: Into<Self::ProjectAction> + Send + Copy + Sync>(
+        &self,
+        metadata: &RequestMetadata,
+        projects_with_actions: &[(&ProjectId, A)],
+    ) -> Result<MustUse<Vec<bool>>, AuthorizationBackendUnavailable> {
+        if metadata.has_admin_privileges() {
+            Ok(vec![true; projects_with_actions.len()])
+        } else {
+            let converted: Vec<(&ProjectId, Self::ProjectAction)> = projects_with_actions
+                .iter()
+                .map(|(id, action)| (*id, (*action).into()))
+                .collect();
+            let decisions = self.are_allowed_project_actions_impl(metadata, &converted)
+                .await;
+
+            #[cfg(debug_assertions)]
+            {
+                if let Ok(ref decisions) = decisions {
+                    assert_eq!(
+                        decisions.len(),
+                        projects_with_actions.len(),
+                        "The number of decisions returned by are_allowed_project_actions_impl does not match the number of project-action pairs provided."
+                    );
+                }
+            }
+
+            decisions
+        }
+        .map(MustUse::from)
+    }
+
     async fn is_allowed_project_action(
         &self,
         metadata: &RequestMetadata,