Authorization checks in OData

src/adapter/odata/src/context.rs

@@ -18,6 +18,7 @@
 
 use std::sync::Arc;
 
+use auth::{DatasetAction, DatasetActionAuthorizer};
 use axum::async_trait;
 use chrono::{DateTime, Utc};
 use datafusion::arrow::datatypes::{Schema, SchemaRef};
@@ -58,7 +59,6 @@ impl ODataServiceContext {
     }
 }
 
-// TODO: Authorization checks
 #[async_trait]
 impl ServiceContext for ODataServiceContext {
     fn service_base_url(&self) -> String {
@@ -69,17 +69,26 @@ impl ServiceContext for ODataServiceContext {
         use futures::TryStreamExt;
 
         let registry: Arc<dyn DatasetRegistry> = self.catalog.get_one().unwrap();
+        let authorizer: Arc<dyn DatasetActionAuthorizer> = self.catalog.get_one().unwrap();
 
-        let datasets = if let Some(account_name) = &self.account_name {
+        let dataset_handles = if let Some(account_name) = &self.account_name {
             registry.get_all_datasets_by_owner(account_name)
         } else {
             registry.get_all_datasets()
         };
 
-        let datasets: Vec<_> = datasets.try_collect().await.unwrap();
+        let dataset_handles: Vec<_> = dataset_handles
+            .try_collect()
+            .await
+            .map_err(ODataError::internal)?;
+
+        let dataset_handles = authorizer
+            .filter_datasets_allowing(dataset_handles, DatasetAction::Read)
+            .await
+            .map_err(ODataError::internal)?;
 
         let mut collections: Vec<Arc<dyn CollectionContext>> = Vec::new();
-        for dataset_handle in datasets {
+        for dataset_handle in dataset_handles {
             let dataset = registry.get_dataset_by_handle(&dataset_handle);
 
             collections.push(Arc::new(ODataCollectionContext {