fix(kafka): take destination tenant from trusted config, not message headers

src/connectors/kafka/config.rs

@@ -82,6 +82,12 @@ pub struct KafkaConfig {
 
 #[derive(Debug, Clone, Args)]
 pub struct ConsumerConfig {
+    /// Trusted destination for every consumed topic. Headers never grant tenant access.
+    #[arg(long = "consumer-tenant-id", env = "P_KAFKA_CONSUMER_TENANT_ID",
+        help = "Destination tenant for all consumed topics; requires --multi-tenancy. Use a separate consumer group or separate topics per tenant. In multi-tenant mode, mismatching x-p-tenant headers are dropped; when unset, all x-p-tenant headers (including DEFAULT_TENANT) are dropped. Single-tenant mode ignores headers. Leave P_KAFKA_CONSUMER_TENANT_ID unset, not empty",
+        value_parser = clap::builder::NonEmptyStringValueParser::new())]
+    pub tenant_id: Option<String>,
+
     #[arg(
         long = "consumer-topics",
         env = "P_KAFKA_CONSUMER_TOPICS",
@@ -585,6 +591,26 @@ pub struct SecurityConfig {
 }
 
 impl KafkaConfig {
+    /// Validate before creating a consumer: tenant IDs become object-store paths.
+    pub fn validate_tenant_mode(&self, multi_tenant: bool) -> anyhow::Result<()> {
+        if let Some(tenant) = self
+            .consumer
+            .as_ref()
+            .and_then(|consumer| consumer.tenant_id.as_deref())
+        {
+            anyhow::ensure!(
+                multi_tenant,
+                "--consumer-tenant-id / P_KAFKA_CONSUMER_TENANT_ID requires --multi-tenancy / P_MULTI_TENANCY"
+            );
+            crate::parseable::validate_tenant_id(tenant).map_err(|error| {
+                anyhow::anyhow!(
+                    "Invalid --consumer-tenant-id / P_KAFKA_CONSUMER_TENANT_ID: {error}"
+                )
+            })?;
+        }
+        Ok(())
+    }
+
     pub fn to_rdkafka_consumer_config(&self) -> ClientConfig {
         let mut config = ClientConfig::new();
 
@@ -1135,6 +1161,7 @@ impl Default for ProducerConfig {
 impl Default for ConsumerConfig {
     fn default() -> Self {
         Self {
+            tenant_id: None,
             topics: vec![],
             group_id: "parseable-connect-cg".to_string(),
             buffer_size: 10_000,
@@ -1511,4 +1538,67 @@ mod tests {
         assert_eq!(config.get("sasl.oauthbearer.client.id"), Some("unused"));
         assert_eq!(config.get("sasl.oauthbearer.client.secret"), Some("unused"));
     }
+
+    #[test]
+    fn empty_consumer_tenant_id_is_rejected_by_cli() {
+        #[derive(Debug, Parser)]
+        struct ConsumerCli {
+            #[command(flatten)]
+            consumer: ConsumerConfig,
+        }
+
+        // A valid value proves this CLI fixture can otherwise parse the option.
+        let cli = ConsumerCli::try_parse_from(["test", "--consumer-tenant-id=tenant-a"])
+            .expect("a nonempty tenant ID should parse");
+        assert_eq!(cli.consumer.tenant_id.as_deref(), Some("tenant-a"));
+        let error = ConsumerCli::try_parse_from(["test", "--consumer-tenant-id="])
+            .expect_err("an explicitly empty tenant ID must fail parsing");
+        assert_eq!(error.kind(), clap::error::ErrorKind::InvalidValue);
+        assert!(error.to_string().contains("--consumer-tenant-id"));
+    }
+}
+
+#[cfg(test)]
+mod tenant_env_tests {
+    use super::*;
+
+    #[test]
+    fn empty_consumer_tenant_env_is_rejected_in_child_process() {
+        let output = std::process::Command::new(std::env::current_exe().unwrap())
+            .args([
+                "--ignored",
+                "--exact",
+                "connectors::kafka::config::tenant_env_tests::tenant_env_child",
+                "--nocapture",
+            ])
+            .env("P_KAFKA_CONSUMER_TENANT_ID", "")
+            .env("PARSEABLE_TENANT_ENV_TEST_CHILD", "1")
+            .output()
+            .unwrap();
+        assert!(
+            output.status.success(),
+            "{}\n{}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+        assert!(String::from_utf8_lossy(&output.stdout).contains("1 passed"));
+    }
+
+    #[test]
+    #[ignore = "run only as a child of empty_consumer_tenant_env_is_rejected_in_child_process"]
+    fn tenant_env_child() {
+        if std::env::var_os("PARSEABLE_TENANT_ENV_TEST_CHILD").is_none() {
+            return;
+        }
+        #[derive(Debug, clap::Parser)]
+        struct ConsumerCli {
+            #[command(flatten)]
+            consumer: ConsumerConfig,
+        }
+        use clap::Parser;
+        let error = ConsumerCli::try_parse_from(["test"])
+            .expect_err("empty tenant env must fail startup parsing");
+        assert_eq!(error.kind(), clap::error::ErrorKind::InvalidValue);
+        assert!(error.to_string().contains("--consumer-tenant-id"));
+    }
 }

src/connectors/kafka/consumer.rs

@@ -24,9 +24,9 @@ use crate::connectors::kafka::{
     ConsumerRecord, KafkaContext, StreamConsumer, TopicPartition, partition_stream,
 };
 use futures_util::FutureExt;
+use rdkafka::Message;
 use rdkafka::Statistics;
 use rdkafka::consumer::Consumer;
-use rdkafka::message::BorrowedMessage;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::{RwLock, mpsc};
@@ -38,6 +38,7 @@ pub struct KafkaStreams {
     stream_state: Arc<RwLock<StreamState>>,
     statistics: Arc<std::sync::RwLock<Statistics>>,
     shutdown_handle: Shutdown,
+    multi_tenant: bool,
 }
 
 impl KafkaStreams {
@@ -47,6 +48,20 @@ impl KafkaStreams {
         shutdown_handle: Shutdown,
     ) -> anyhow::Result<KafkaStreams> {
         info!("Initializing KafkaStreams...");
+        let multi_tenant = crate::parseable::PARSEABLE.options.is_multi_tenant();
+        context.config.validate_tenant_mode(multi_tenant)?;
+        if multi_tenant
+            && context
+                .config
+                .consumer
+                .as_ref()
+                .and_then(|config| config.tenant_id.as_ref())
+                .is_none()
+        {
+            warn!(
+                "Multi-tenant Kafka consumer has no destination tenant; set --consumer-tenant-id / P_KAFKA_CONSUMER_TENANT_ID. Headerless records use the default destination; records carrying tenant headers are dropped"
+            );
+        }
         let statistics = Arc::clone(&context.statistics);
         let consumer = KafkaStreams::create_consumer(context);
         info!("KafkaStreams initialized successfully.");
@@ -56,6 +71,7 @@ impl KafkaStreams {
             stream_state,
             statistics,
             shutdown_handle,
+            multi_tenant,
         })
     }
 
@@ -102,6 +118,13 @@ impl KafkaStreams {
     pub fn partitioned(&self) -> ReceiverStream<PartitionStreamReceiver> {
         let (stream_tx, stream_rx) = mpsc::channel(100);
         let consumer = self.consumer();
+        let multi_tenant = self.multi_tenant;
+        let configured_tenant = consumer
+            .context()
+            .config
+            .consumer
+            .as_ref()
+            .and_then(|config| config.tenant_id.clone());
         let stream_state = self.state();
         let tokio_handle = tokio::runtime::Handle::current();
         let shutdown_handle = self.shutdown_handle.clone();
@@ -112,7 +135,7 @@ impl KafkaStreams {
                     let result: Result<(), ConnectorError> = tokio::select! {
                         result = consumer.recv() => {
                             match result {
-                                Ok(msg) => KafkaStreams::handle_message(msg, &stream_state, &stream_tx).await.map_err(Into::into),
+                                Ok(msg) => KafkaStreams::handle_message(msg, &stream_state, &stream_tx, configured_tenant.as_deref(), multi_tenant).await.map_err(Into::into),
                                 Err(e) => Err(e.into())
                             }
                         }
@@ -152,14 +175,16 @@ impl KafkaStreams {
     }
 
     /// Handle individual Kafka message and route it to the proper partition stream
-    async fn handle_message(
-        msg: BorrowedMessage<'_>,
+    async fn handle_message<M: Message>(
+        msg: M,
         stream_state: &RwLock<StreamState>,
         stream_tx: &mpsc::Sender<PartitionStreamReceiver>,
+        configured_tenant: Option<&str>,
+        multi_tenant: bool,
     ) -> anyhow::Result<()> {
+        let tp = TopicPartition::new(msg.topic().to_owned(), msg.partition());
+        let consumer_record = ConsumerRecord::from_message(msg, configured_tenant, multi_tenant);
         let mut state = stream_state.write().await;
-        let tp = TopicPartition::from_kafka_msg(&msg);
-        let consumer_record = ConsumerRecord::from_borrowed_msg(msg);
 
         let partition_stream_tx =
             KafkaStreams::get_or_create_partition_stream(&mut state, stream_tx, tp).await;
@@ -244,3 +269,52 @@ impl KafkaStreams {
         };
     }
 }
+
+#[cfg(test)]
+mod routing_tests {
+    use super::*;
+    use futures_util::StreamExt;
+
+    #[tokio::test]
+    async fn rejected_record_is_delivered_to_partition_stream_receiver() {
+        let state = RwLock::new(StreamState::new(1));
+        let (tx, mut rx) = mpsc::channel(1);
+        KafkaStreams::handle_message(
+            crate::connectors::kafka::message_conversion_tests::message("tenant-b"),
+            &state,
+            &tx,
+            Some("tenant-a"),
+            true,
+        )
+        .await
+        .unwrap();
+        let partition = tokio::time::timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .expect("rejected record must create a partition stream")
+            .unwrap();
+        assert_eq!(
+            partition.topic_partition(),
+            &TopicPartition::new("trusted-topic".into(), 3)
+        );
+        partition
+            .run_drain(|mut records| async move {
+                let record = tokio::time::timeout(Duration::from_secs(1), records.next())
+                    .await
+                    .expect("marker must be routed to the partition receiver")
+                    .unwrap();
+                assert_eq!(
+                    record,
+                    ConsumerRecord::new(
+                        None,
+                        None,
+                        "trusted-topic".into(),
+                        3,
+                        42,
+                        Some(123456),
+                        Some("tenant-a".into()),
+                    )
+                );
+            })
+            .await;
+    }
+}

src/connectors/kafka/metrics.rs

@@ -24,6 +24,16 @@ use prometheus::{
 use rdkafka::Statistics;
 use std::sync::{Arc, RwLock};
 
+/// Rejected records, without attacker-controlled labels.
+pub static TENANT_MISMATCH_DROPPED: once_cell::sync::Lazy<prometheus::IntCounter> =
+    once_cell::sync::Lazy::new(|| {
+        prometheus::IntCounter::new(
+            "parseable_kafka_tenant_mismatch_dropped_total",
+            "Kafka records dropped because a tenant header disagrees with trusted configuration",
+        )
+        .expect("valid Kafka tenant counter")
+    });
+
 #[derive(Debug)]
 pub struct KafkaMetricsCollector {
     stats: Arc<RwLock<Statistics>>,