fix: auth flow for query and permission assignment for ListStream (#1048)

---------
Co-authored-by: Nikhil Sinha <[email protected]>

Author: parmesant

src/correlation/correlation_utils.rs

@@ -18,9 +18,8 @@
 
 use std::collections::HashSet;
 
-use actix_web::http::header::ContentType;
+use actix_web::{http::header::ContentType, Error};
 use chrono::Utc;
-use correlation_utils::user_auth_for_query;
 use datafusion::error::DataFusionError;
 use http::StatusCode;
 use itertools::Itertools;
@@ -31,12 +30,15 @@ use tokio::sync::RwLock;
 use tracing::{error, trace, warn};
 
 use crate::{
-    handlers::http::rbac::RBACError, option::CONFIG, query::QUERY_SESSION, rbac::map::SessionKey,
-    storage::ObjectStorageError, users::filters::FilterQuery, utils::get_hash,
+    handlers::http::rbac::RBACError,
+    option::CONFIG,
+    query::QUERY_SESSION,
+    rbac::{map::SessionKey, Users},
+    storage::ObjectStorageError,
+    users::filters::FilterQuery,
+    utils::{get_hash, user_auth_for_query},
 };
 
-pub mod correlation_utils;
-
 pub static CORRELATIONS: Lazy<Correlation> = Lazy::new(Correlation::default);
 
 #[derive(Debug, Default)]
@@ -77,11 +79,15 @@ impl Correlation {
         let correlations = self.0.read().await.iter().cloned().collect_vec();
 
         let mut user_correlations = vec![];
+        let permissions = Users.get_permissions(session_key);
+
         for c in correlations {
-            if user_auth_for_query(session_key, &c.table_configs)
-                .await
-                .is_ok()
-            {
+            let tables = &c
+                .table_configs
+                .iter()
+                .map(|t| t.table_name.clone())
+                .collect_vec();
+            if user_auth_for_query(&permissions, tables).is_ok() {
                 user_correlations.push(c);
             }
         }
@@ -220,7 +226,14 @@ impl CorrelationRequest {
         }
 
         // check if user has access to table
-        user_auth_for_query(session_key, &self.table_configs).await?;
+        let permissions = Users.get_permissions(session_key);
+        let tables = &self
+            .table_configs
+            .iter()
+            .map(|t| t.table_name.clone())
+            .collect_vec();
+
+        user_auth_for_query(&permissions, tables)?;
 
         // to validate table config, we need to check whether the mentioned fields
         // are present in the table or not
@@ -271,6 +284,8 @@ pub enum CorrelationError {
     Unauthorized,
     #[error("DataFusion Error: {0}")]
     DataFusion(#[from] DataFusionError),
+    #[error("{0}")]
+    ActixError(#[from] Error),
 }
 
 impl actix_web::ResponseError for CorrelationError {
@@ -283,6 +298,7 @@ impl actix_web::ResponseError for CorrelationError {
             Self::AnyhowError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             Self::Unauthorized => StatusCode::BAD_REQUEST,
             Self::DataFusion(_) => StatusCode::INTERNAL_SERVER_ERROR,
+            Self::ActixError(_) => StatusCode::BAD_REQUEST,
         }
     }
 

src/correlation/mod.rs

@@ -34,9 +34,7 @@ use tonic::transport::{Identity, Server, ServerTlsConfig};
 use tonic_web::GrpcWebLayer;
 
 use crate::handlers::http::cluster::get_ingestor_info;
-use crate::handlers::http::query::{
-    authorize_and_set_filter_tags, into_query, update_schema_when_distributed,
-};
+use crate::handlers::http::query::{into_query, update_schema_when_distributed};
 use crate::handlers::livetail::cross_origin_config;
 use crate::metrics::QUERY_EXECUTE_TIME;
 use crate::option::CONFIG;
@@ -46,6 +44,7 @@ use crate::utils::arrow::flight::{
     send_to_ingester,
 };
 use crate::utils::time::TimeRange;
+use crate::utils::user_auth_for_query;
 use arrow_flight::{
     flight_service_server::FlightService, Action, ActionType, Criteria, Empty, FlightData,
     FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse, PutResult, SchemaAsIpc,
@@ -157,12 +156,12 @@ impl FlightService for AirServiceImpl {
             .ok_or_else(|| Status::aborted("Malformed SQL Provided, Table Name Not Found"))?
             .to_owned();
 
-        update_schema_when_distributed(streams)
+        update_schema_when_distributed(&streams)
             .await
             .map_err(|err| Status::internal(err.to_string()))?;
 
         // map payload to query
-        let mut query = into_query(&ticket, &session_state, time_range)
+        let query = into_query(&ticket, &session_state, time_range)
             .await
             .map_err(|_| Status::internal("Failed to parse query"))?;
 
@@ -202,7 +201,7 @@ impl FlightService for AirServiceImpl {
             rbac::Response::Authorized => (),
             rbac::Response::UnAuthorized => {
                 return Err(Status::permission_denied(
-                    "user is not authenticated to access this resource",
+                    "user is not authorized to access this resource",
                 ))
             }
             rbac::Response::ReloadRequired => {
@@ -212,7 +211,7 @@ impl FlightService for AirServiceImpl {
 
         let permissions = Users.get_permissions(&key);
 
-        authorize_and_set_filter_tags(&mut query, permissions, &stream_name).map_err(|_| {
+        user_auth_for_query(&permissions, &streams).map_err(|_| {
             Status::permission_denied("User Does not have permission to access this")
         })?;
         let time = Instant::now();

src/handlers/airplane.rs

@@ -18,17 +18,17 @@
 
 use actix_web::{web, HttpRequest, Responder};
 use bytes::Bytes;
+use itertools::Itertools;
 use relative_path::RelativePathBuf;
 
+use crate::rbac::Users;
+use crate::utils::user_auth_for_query;
 use crate::{
     option::CONFIG, storage::CORRELATIONS_ROOT_DIRECTORY,
     utils::actix::extract_session_key_from_req,
 };
 
-use crate::correlation::{
-    correlation_utils::user_auth_for_query, CorrelationConfig, CorrelationError,
-    CorrelationRequest, CORRELATIONS,
-};
+use crate::correlation::{CorrelationConfig, CorrelationError, CorrelationRequest, CORRELATIONS};
 
 pub async fn list(req: HttpRequest) -> Result<impl Responder, CorrelationError> {
     let session_key = extract_session_key_from_req(&req)
@@ -52,14 +52,17 @@ pub async fn get(req: HttpRequest) -> Result<impl Responder, CorrelationError> {
 
     let correlation = CORRELATIONS.get_correlation_by_id(correlation_id).await?;
 
-    if user_auth_for_query(&session_key, &correlation.table_configs)
-        .await
-        .is_ok()
-    {
-        Ok(web::Json(correlation))
-    } else {
-        Err(CorrelationError::Unauthorized)
-    }
+    let permissions = Users.get_permissions(&session_key);
+
+    let tables = &correlation
+        .table_configs
+        .iter()
+        .map(|t| t.table_name.clone())
+        .collect_vec();
+
+    user_auth_for_query(&permissions, tables)?;
+
+    Ok(web::Json(correlation))
 }
 
 pub async fn post(req: HttpRequest, body: Bytes) -> Result<impl Responder, CorrelationError> {
@@ -93,7 +96,14 @@ pub async fn modify(req: HttpRequest, body: Bytes) -> Result<impl Responder, Cor
 
     // validate whether user has access to this correlation object or not
     let correlation = CORRELATIONS.get_correlation_by_id(correlation_id).await?;
-    user_auth_for_query(&session_key, &correlation.table_configs).await?;
+    let permissions = Users.get_permissions(&session_key);
+    let tables = &correlation
+        .table_configs
+        .iter()
+        .map(|t| t.table_name.clone())
+        .collect_vec();
+
+    user_auth_for_query(&permissions, tables)?;
 
     let correlation_request: CorrelationRequest = serde_json::from_slice(&body)?;
     correlation_request.validate(&session_key).await?;
@@ -122,7 +132,14 @@ pub async fn delete(req: HttpRequest) -> Result<impl Responder, CorrelationError
     let correlation = CORRELATIONS.get_correlation_by_id(correlation_id).await?;
 
     // validate user's query auth
-    user_auth_for_query(&session_key, &correlation.table_configs).await?;
+    let permissions = Users.get_permissions(&session_key);
+    let tables = &correlation
+        .table_configs
+        .iter()
+        .map(|t| t.table_name.clone())
+        .collect_vec();
+
+    user_auth_for_query(&permissions, tables)?;
 
     // Delete from disk
     let store = CONFIG.storage().get_object_store();

src/handlers/http/correlation.rs

@@ -32,9 +32,12 @@ use crate::hottier::{HotTierManager, StreamHotTier, CURRENT_HOT_TIER_VERSION};
 use crate::metadata::{SchemaVersion, STREAM_INFO};
 use crate::metrics::{EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE};
 use crate::option::{Mode, CONFIG};
+use crate::rbac::role::Action;
+use crate::rbac::Users;
 use crate::stats::{event_labels_date, storage_size_labels_date, Stats};
 use crate::storage::StreamType;
 use crate::storage::{retention::Retention, StorageDir, StreamInfo};
+use crate::utils::actix::extract_session_key_from_req;
 use crate::{event, stats};
 
 use crate::{metadata, validator};
@@ -46,6 +49,7 @@ use arrow_schema::{Field, Schema};
 use bytes::Bytes;
 use chrono::Utc;
 use http::{HeaderName, HeaderValue};
+use itertools::Itertools;
 use serde_json::Value;
 use std::collections::HashMap;
 use std::fs;
@@ -86,16 +90,27 @@ pub async fn delete(req: HttpRequest) -> Result<impl Responder, StreamError> {
     Ok((format!("log stream {stream_name} deleted"), StatusCode::OK))
 }
 
-pub async fn list(_: HttpRequest) -> impl Responder {
-    //list all streams from storage
+pub async fn list(req: HttpRequest) -> Result<impl Responder, StreamError> {
+    let key = extract_session_key_from_req(&req)
+        .map_err(|err| StreamError::Anyhow(anyhow::Error::msg(err.to_string())))?;
+
+    // list all streams from storage
     let res = CONFIG
         .storage()
         .get_object_store()
         .list_streams()
         .await
-        .unwrap();
+        .unwrap()
+        .into_iter()
+        .filter(|logstream| {
+            warn!("logstream-\n{logstream:?}");
+
+            Users.authorize(key.clone(), Action::ListStream, Some(&logstream.name), None)
+                == crate::rbac::Response::Authorized
+        })
+        .collect_vec();
 
-    web::Json(res)
+    Ok(web::Json(res))
 }
 
 pub async fn detect_schema(body: Bytes) -> Result<impl Responder, StreamError> {
@@ -130,7 +145,7 @@ pub async fn schema(req: HttpRequest) -> Result<impl Responder, StreamError> {
         }
         Err(err) => return Err(StreamError::from(err)),
     };
-    match update_schema_when_distributed(vec![stream_name.clone()]).await {
+    match update_schema_when_distributed(&vec![stream_name.clone()]).await {
         Ok(_) => {
             let schema = STREAM_INFO.schema(&stream_name)?;
             Ok((web::Json(schema), StatusCode::OK))