Feature: Backend for Correlation

src/cli.rs

@@ -529,7 +529,9 @@ impl FromArgMatches for Cli {
         self.kafka_host = m.get_one::<String>(Self::KAFKA_HOST).cloned();
         self.kafka_group = m.get_one::<String>(Self::KAFKA_GROUP).cloned();
         self.kafka_client_id = m.get_one::<String>(Self::KAFKA_CLIENT_ID).cloned();
-        self.kafka_security_protocol = m.get_one::<SslProtocol>(Self::KAFKA_SECURITY_PROTOCOL).cloned();
+        self.kafka_security_protocol = m
+            .get_one::<SslProtocol>(Self::KAFKA_SECURITY_PROTOCOL)
+            .cloned();
         self.kafka_partitions = m.get_one::<String>(Self::KAFKA_PARTITIONS).cloned();
 
         self.tls_cert_path = m.get_one::<PathBuf>(Self::TLS_CERT).cloned();

src/correlation/correlation_utils.rs

@@ -0,0 +1,77 @@
+/*
+ * Parseable Server (C) 2022 - 2024 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+use datafusion::common::tree_node::TreeNode;
+
+use crate::{
+    query::{TableScanVisitor, QUERY_SESSION},
+    rbac::{
+        map::SessionKey,
+        role::{Action, Permission},
+        Users,
+    },
+};
+
+use super::CorrelationError;
+
+async fn get_tables_from_query(query: &str) -> Result<TableScanVisitor, CorrelationError> {
+    let session_state = QUERY_SESSION.state();
+    let raw_logical_plan = session_state
+        .create_logical_plan(query)
+        .await
+        .map_err(|err| CorrelationError::AnyhowError(err.into()))?;
+
+    let mut visitor = TableScanVisitor::default();
+    let _ = raw_logical_plan.visit(&mut visitor);
+    Ok(visitor)
+}
+
+pub async fn user_auth_for_query(
+    session_key: &SessionKey,
+    query: &str,
+) -> Result<(), CorrelationError> {
+    let tables = get_tables_from_query(query).await?;
+    let permissions = Users.get_permissions(session_key);
+
+    for table_name in tables.into_inner().iter() {
+        let mut authorized = false;
+
+        // in permission check if user can run query on the stream.
+        // also while iterating add any filter tags for this stream
+        for permission in permissions.iter() {
+            match permission {
+                Permission::Stream(Action::All, _) => {
+                    authorized = true;
+                    break;
+                }
+                Permission::StreamWithTag(Action::Query, ref stream, _)
+                    if stream == table_name || stream == "*" =>
+                {
+                    authorized = true;
+                }
+                _ => (),
+            }
+        }
+
+        if !authorized {
+            return Err(CorrelationError::Unauthorized);
+        }
+    }
+
+    Ok(())
+}

src/correlation/http_handlers.rs

@@ -0,0 +1,149 @@
+/*
+ * Parseable Server (C) 2022 - 2024 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+use actix_web::{web, HttpRequest, Responder};
+use bytes::Bytes;
+use relative_path::RelativePathBuf;
+
+use crate::{
+    option::CONFIG,
+    storage::{CORRELATION_DIRECTORY, PARSEABLE_ROOT_DIRECTORY},
+    utils::{actix::extract_session_key_from_req, uid::Uid},
+};
+
+use super::{
+    correlation_utils::user_auth_for_query, CorrelationConfig, CorrelationError,
+    CorrelationRequest, CORRELATIONS,
+};
+
+pub async fn list(req: HttpRequest) -> Result<impl Responder, CorrelationError> {
+    let session_key = extract_session_key_from_req(&req)
+        .map_err(|err| CorrelationError::AnyhowError(anyhow::Error::msg(err.to_string())))?;
+
+    let correlations = CORRELATIONS
+        .list_correlations_for_user(&session_key)
+        .await?;
+
+    Ok(web::Json(correlations))
+}
+
+pub async fn get(req: HttpRequest) -> Result<impl Responder, CorrelationError> {
+    let session_key = extract_session_key_from_req(&req)
+        .map_err(|err| CorrelationError::AnyhowError(anyhow::Error::msg(err.to_string())))?;
+
+    let correlation_id = req
+        .match_info()
+        .get("correlation_id")
+        .ok_or(CorrelationError::Metadata("No correlation ID Provided"))?;
+
+    let correlation = CORRELATIONS.get_correlation_by_id(correlation_id).await?;
+
+    if user_auth_for_query(&session_key, &correlation.query)
+        .await
+        .is_ok()
+    {
+        Ok(web::Json(correlation))
+    } else {
+        Err(CorrelationError::Unauthorized)
+    }
+}
+
+pub async fn post(req: HttpRequest, body: Bytes) -> Result<impl Responder, CorrelationError> {
+    let session_key = extract_session_key_from_req(&req)
+        .map_err(|err| CorrelationError::AnyhowError(anyhow::Error::msg(err.to_string())))?;
+
+    let correlation_request: CorrelationRequest = serde_json::from_slice(&body)?;
+    let correlation: CorrelationConfig = correlation_request.into();
+
+    // validate user's query auth
+    user_auth_for_query(&session_key, &correlation.query).await?;
+
+    // Save to disk
+    let store = CONFIG.storage().get_object_store();
+    store.put_correlation(&correlation).await?;
+
+    // Save to memory
+    CORRELATIONS.update(&correlation).await?;
+
+    Ok(format!(
+        "Saved correlation with ID- {}",
+        correlation.id.to_string()
+    ))
+}
+
+pub async fn modify(req: HttpRequest, body: Bytes) -> Result<impl Responder, CorrelationError> {
+    let session_key = extract_session_key_from_req(&req)
+        .map_err(|err| CorrelationError::AnyhowError(anyhow::Error::msg(err.to_string())))?;
+
+    let correlation_id = req
+        .match_info()
+        .get("correlation_id")
+        .ok_or(CorrelationError::Metadata("No correlation ID Provided"))?;
+
+    let correlation_request: CorrelationRequest = serde_json::from_slice(&body)?;
+
+    // validate user's query auth
+    user_auth_for_query(&session_key, &correlation_request.query).await?;
+
+    let correlation: CorrelationConfig = CorrelationConfig {
+        version: correlation_request.version,
+        id: Uid::from_string(correlation_id)
+            .map_err(|err| CorrelationError::AnyhowError(anyhow::Error::msg(err.to_string())))?,
+        query: correlation_request.query,
+    };
+
+    // Save to disk
+    let store = CONFIG.storage().get_object_store();
+    store.put_correlation(&correlation).await?;
+
+    // Save to memory
+    CORRELATIONS.update(&correlation).await?;
+
+    Ok(format!(
+        "Modified correlation with ID- {}",
+        correlation.id.to_string()
+    ))
+}
+
+pub async fn delete(req: HttpRequest) -> Result<impl Responder, CorrelationError> {
+    let session_key = extract_session_key_from_req(&req)
+        .map_err(|err| CorrelationError::AnyhowError(anyhow::Error::msg(err.to_string())))?;
+
+    let correlation_id = req
+        .match_info()
+        .get("correlation_id")
+        .ok_or(CorrelationError::Metadata("No correlation ID Provided"))?;
+
+    let correlation = CORRELATIONS.get_correlation_by_id(correlation_id).await?;
+
+    // validate user's query auth
+    user_auth_for_query(&session_key, &correlation.query).await?;
+
+    // Delete from disk
+    let store = CONFIG.storage().get_object_store();
+    let path = RelativePathBuf::from_iter([
+        PARSEABLE_ROOT_DIRECTORY,
+        CORRELATION_DIRECTORY,
+        &format!("{}", correlation.id),
+    ]);
+    store.delete_object(&path).await?;
+
+    // Delete from memory
+    CORRELATIONS.delete(correlation_id).await?;
+    Ok(format!("Deleted correlation with ID- {correlation_id}"))
+}