add validation for otel datasets

Author: nikhilsinhaparseable

src/handlers/http/datasets.rs

@@ -22,13 +22,24 @@ use actix_web::http::StatusCode;
 use actix_web::{HttpRequest, HttpResponse, web};
 use serde::{Deserialize, Serialize};
 
+use crate::rbac::{self, Users, role::Action};
+use crate::utils::actix::extract_session_key_from_req;
 use crate::utils::get_tenant_id_from_request;
 use crate::{
     handlers::DatasetTag,
     parseable::PARSEABLE,
     storage::{ObjectStorageError, StreamType},
 };
 
+/// Check if the caller is authorized to read a specific stream.
+fn can_access_stream(req: &HttpRequest, stream_name: &str) -> bool {
+    let Ok(key) = extract_session_key_from_req(req) else {
+        return false;
+    };
+    Users.authorize(key, Action::GetStreamInfo, Some(stream_name), None)
+        == rbac::Response::Authorized
+}
+
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct CorrelatedDataset {
@@ -39,12 +50,19 @@ struct CorrelatedDataset {
 
 /// GET /api/v1/datasets/{name}/correlated
 /// Returns all datasets sharing at least one tag or label with the named dataset.
+/// Results are filtered to only include datasets the caller is authorized to read.
 pub async fn get_correlated_datasets(
     req: HttpRequest,
     path: web::Path<String>,
 ) -> Result<HttpResponse, DatasetsError> {
     let dataset_name = path.into_inner();
     let tenant_id = get_tenant_id_from_request(&req);
+
+    // Authorize caller for the seed dataset
+    if !can_access_stream(&req, &dataset_name) {
+        return Err(DatasetsError::DatasetNotFound(dataset_name));
+    }
+
     let stream = PARSEABLE
         .get_stream(&dataset_name, &tenant_id)
         .map_err(|_| DatasetsError::DatasetNotFound(dataset_name.clone()))?;
@@ -63,6 +81,10 @@ pub async fn get_correlated_datasets(
         if name == dataset_name {
             continue;
         }
+        // Filter out datasets the caller cannot read
+        if !can_access_stream(&req, &name) {
+            continue;
+        }
         if let Ok(s) = PARSEABLE.get_stream(&name, &tenant_id) {
             // Skip internal streams
             if s.get_stream_type() == StreamType::Internal {
@@ -91,6 +113,7 @@ pub async fn get_correlated_datasets(
 
 /// GET /api/v1/datasets/tags/{tag}
 /// Returns all datasets that have the specified tag.
+/// Results are filtered to only include datasets the caller is authorized to read.
 pub async fn get_datasets_by_tag(
     req: HttpRequest,
     path: web::Path<String>,
@@ -104,6 +127,10 @@ pub async fn get_datasets_by_tag(
     let mut matching = Vec::new();
 
     for name in all_streams {
+        // Filter out datasets the caller cannot read
+        if !can_access_stream(&req, &name) {
+            continue;
+        }
         if let Ok(s) = PARSEABLE.get_stream(&name, &tenant_id) {
             if s.get_stream_type() == StreamType::Internal {
                 continue;

src/handlers/http/ingest.rs

@@ -32,7 +32,7 @@ use crate::event::format::{self, EventFormat, LogSource, LogSourceEntry};
 use crate::event::{self, FORMAT_KEY, USER_AGENT_KEY};
 use crate::handlers::http::modal::utils::ingest_utils::validate_stream_for_ingestion;
 use crate::handlers::{
-    CONTENT_TYPE_JSON, CONTENT_TYPE_PROTOBUF, EXTRACT_LOG_KEY, LOG_SOURCE_KEY,
+    CONTENT_TYPE_JSON, CONTENT_TYPE_PROTOBUF, DatasetTag, EXTRACT_LOG_KEY, LOG_SOURCE_KEY,
     STREAM_NAME_HEADER_KEY, TELEMETRY_TYPE_KEY, TelemetryType,
 };
 use crate::metadata::SchemaVersion;
@@ -207,6 +207,8 @@ pub async fn setup_otel_stream(
     expected_log_source: LogSource,
     known_fields: &[&str],
     telemetry_type: TelemetryType,
+    dataset_tags: Vec<DatasetTag>,
+    dataset_labels: Vec<String>,
 ) -> Result<(String, LogSource, LogSourceEntry, Option<String>), PostError> {
     let Some(stream_name) = req.headers().get(STREAM_NAME_HEADER_KEY) else {
         return Err(PostError::Header(ParseHeaderError::MissingStreamName));
@@ -240,8 +242,8 @@ pub async fn setup_otel_stream(
             vec![log_source_entry.clone()],
             telemetry_type,
             &tenant_id,
-            vec![],
-            vec![],
+            dataset_tags,
+            dataset_labels,
         )
         .await?;
     let mut time_partition = None;
@@ -364,6 +366,8 @@ pub async fn handle_otel_logs_ingestion(
         LogSource::OtelLogs,
         &OTEL_LOG_KNOWN_FIELD_LIST,
         TelemetryType::Logs,
+        vec![],
+        vec![],
     )
     .await
     .map_err(|e| {
@@ -388,6 +392,8 @@ pub async fn handle_otel_metrics_ingestion(
         LogSource::OtelMetrics,
         &OTEL_METRICS_KNOWN_FIELD_LIST,
         TelemetryType::Metrics,
+        vec![],
+        vec![],
     )
     .await
     .map_err(|e| {
@@ -419,6 +425,8 @@ pub async fn handle_otel_traces_ingestion(
         LogSource::OtelTraces,
         &OTEL_TRACES_KNOWN_FIELD_LIST,
         TelemetryType::Traces,
+        vec![],
+        vec![],
     )
     .await
     .map_err(|e| {

src/handlers/http/modal/server.rs

@@ -213,13 +213,13 @@ impl Server {
                 "/tags/{tag}",
                 web::get()
                     .to(http::datasets::get_datasets_by_tag)
-                    .authorize_for_resource(Action::GetStreamInfo),
+                    .authorize(Action::GetStreamInfo),
             )
             .route(
                 "/{name}/correlated",
                 web::get()
                     .to(http::datasets::get_correlated_datasets)
-                    .authorize_for_resource(Action::GetStreamInfo),
+                    .authorize(Action::GetStreamInfo),
             )
             .route(
                 "/{name}",

src/handlers/mod.rs

@@ -92,10 +92,10 @@ impl Display for TelemetryType {
 #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
 #[serde(rename_all = "kebab-case")]
 pub enum DatasetTag {
-    AgentMonitoring,
-    K8sMonitoring,
+    AgentObservability,
+    K8sObservability,
     DatabaseObservability,
-    ApplicationMonitoring,
+    APM,
     ServiceMap,
 }
 
@@ -104,13 +104,13 @@ impl TryFrom<&str> for DatasetTag {
 
     fn try_from(s: &str) -> Result<Self, Self::Error> {
         match s.to_lowercase().as_str() {
-            "agent-monitoring" => Ok(DatasetTag::AgentMonitoring),
-            "k8s-monitoring" => Ok(DatasetTag::K8sMonitoring),
+            "agent-observability" => Ok(DatasetTag::AgentObservability),
+            "k8s-observability" => Ok(DatasetTag::K8sObservability),
             "database-observability" => Ok(DatasetTag::DatabaseObservability),
-            "application-monitoring" => Ok(DatasetTag::ApplicationMonitoring),
+            "apm" => Ok(DatasetTag::APM),
             "service-map" => Ok(DatasetTag::ServiceMap),
             _ => Err(
-                "Invalid dataset tag. Supported values: agent-monitoring, k8s-monitoring, database-observability, application-monitoring, service-map",
+                "Invalid dataset tag. Supported values: agent-observability, k8s-observability, database-observability, apm, service-map",
             ),
         }
     }
@@ -119,10 +119,10 @@ impl TryFrom<&str> for DatasetTag {
 impl Display for DatasetTag {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str(match self {
-            DatasetTag::AgentMonitoring => "agent-monitoring",
-            DatasetTag::K8sMonitoring => "k8s-monitoring",
+            DatasetTag::AgentObservability => "agent-observability",
+            DatasetTag::K8sObservability => "k8s-observability",
             DatasetTag::DatabaseObservability => "database-observability",
-            DatasetTag::ApplicationMonitoring => "application-monitoring",
+            DatasetTag::APM => "apm",
             DatasetTag::ServiceMap => "service-map",
         })
     }

src/parseable/mod.rs

@@ -477,7 +477,7 @@ impl Parseable {
                     TelemetryType::Logs,
                     &tenant_id,
                     vec![],
-                    vec![]
+                    vec![],
                 )
                 .await;