implement SASL authentication

Author: matthieugouel

Cargo.lock

@@ -16,5 +16,5 @@ ipnet = "2.10.1"
 log = "0.4.22"
 pcap = "2.2.0"
 rand = "0.8.5"
-rdkafka = { version = "0.37.0" }
+rdkafka = { version = "0.37.0", features = ["sasl", "ssl"] }
 tokio = { version = "1.42.0", features = ["full"] }

Cargo.toml

@@ -10,7 +10,7 @@ use std::time::Duration;
 use tokio::task;
 
 use crate::prober::probe;
-use crate::producer::produce;
+use crate::producer::{produce, KafkaAuth};
 
 /// Probing configuration.
 #[derive(Debug)]
@@ -143,6 +143,7 @@ pub async fn handle(
     _in_topics: &str,
     _in_group_id: &str,
     out_topic: &str,
+    out_auth: KafkaAuth,
     target: &str,
 ) -> Result<()> {
     // Probe Generation
@@ -155,7 +156,7 @@ pub async fn handle(
     let (_, _, results) = result?;
 
     // Produce the results to Kafka topic
-    produce(brokers, out_topic, prober_id, results).await;
+    produce(brokers, out_topic, prober_id, out_auth, results).await;
 
     Ok(())
 }

src/handler.rs

@@ -33,9 +33,25 @@ struct Cli {
     in_group_id: String,
 
     /// Kafka producer topic
-    #[clap(long, default_value = "results")]
+    #[clap(long, default_value = "osiris-results")]
     out_topic: String,
 
+    /// Kafka producer Authentication Protocol
+    #[clap(long, default_value = "PLAINTEXT")]
+    out_auth_protocol: String,
+
+    /// Kafka producer Authentication SASL Username
+    #[clap(long, default_value = "osiris")]
+    out_auth_sasl_username: String,
+
+    /// Kafka producer Authentication SASL Password
+    #[clap(long, default_value = "osiris")]
+    out_auth_sasl_password: String,
+
+    /// Kafka producer Authentication SASL Mechanism
+    #[clap(long, default_value = "SCRAM-SHA-512")]
+    out_auth_sasl_mechanism: String,
+
     /// Target (eg., 2606:4700:4700::1111/128,1,32,1)
     #[arg(index = 1)]
     target: String,
@@ -65,12 +81,28 @@ async fn main() -> Result<()> {
     let cli = Cli::parse();
     set_logging(&cli);
 
+    // Configure Kafka producer authentication
+    let out_auth = match cli.out_auth_protocol.as_str() {
+        "PLAINTEXT" => producer::KafkaAuth::PLAINTEXT,
+        "SASL" => producer::KafkaAuth::SASL(producer::SaslAuth {
+            username: cli.out_auth_sasl_username.clone(),
+            password: cli.out_auth_sasl_password.clone(),
+            mechanism: cli.out_auth_sasl_mechanism.clone(),
+        }),
+        _ => {
+            return Err(anyhow::anyhow!(
+                "Invalid Kafka producer authentication protocol"
+            ))
+        }
+    };
+
     match handle(
         cli.prober_id,
         &cli.brokers,
         &cli.in_topics,
         &cli.in_group_id,
         &cli.out_topic,
+        out_auth,
         &cli.target,
     )
     .await

src/main.rs

@@ -1,10 +1,22 @@
 use caracat::models::{MPLSLabel, Reply};
+use log::info;
 use rdkafka::config::ClientConfig;
 use rdkafka::message::{Header, OwnedHeaders};
 use rdkafka::producer::{FutureProducer, FutureRecord};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
+pub struct SaslAuth {
+    pub username: String,
+    pub password: String,
+    pub mechanism: String,
+}
+
+pub enum KafkaAuth {
+    SASL(SaslAuth),
+    PLAINTEXT,
+}
+
 fn format_mpls_labels(mpls_labels: &Vec<MPLSLabel>) -> String {
     String::from("[")
         + &mpls_labels
@@ -51,16 +63,28 @@ pub async fn produce(
     brokers: &str,
     topic_name: &str,
     prober_id: u16,
+    auth: KafkaAuth,
     results: Arc<Mutex<Vec<Reply>>>,
 ) {
-    let producer: &FutureProducer = &ClientConfig::new()
-        .set("bootstrap.servers", brokers)
-        .set("message.timeout.ms", "5000")
-        .create()
-        .expect("Producer creation error");
+    let producer: &FutureProducer = match auth {
+        KafkaAuth::PLAINTEXT => &ClientConfig::new()
+            .set("bootstrap.servers", brokers)
+            .set("message.timeout.ms", "5000")
+            .create()
+            .expect("Producer creation error"),
+        KafkaAuth::SASL(scram_auth) => &ClientConfig::new()
+            .set("bootstrap.servers", brokers)
+            .set("message.timeout.ms", "5000")
+            .set("sasl.username", scram_auth.username)
+            .set("sasl.password", scram_auth.password)
+            .set("sasl.mechanisms", scram_auth.mechanism)
+            .set("security.protocol", "SASL_PLAINTEXT")
+            .create()
+            .expect("Producer creation error"),
+    };
 
     for result in results.lock().unwrap().iter() {
-        let _ = producer
+        let delivery_status = producer
             .send(
                 FutureRecord::to(topic_name)
                     .payload(&format!("{}", format_reply(prober_id, result)))
@@ -72,6 +96,6 @@ pub async fn produce(
                 Duration::from_secs(0),
             )
             .await;
+        info!("{:?}", delivery_status);
     }
-    // let delivery_status = info!("Delivery status: {:?}", delivery_status);
 }

src/producer.rs

@@ -2,8 +2,9 @@
 
 Docker Compose setup to facilitate the tests of Osiris.
 
-The testbed consists in a Redpanda and ClickHouse instance. Required ClickHouse [tables](config/clickhouse/docker-entrypoint-initdb.d/init.sql) are created on startup. The `osiris.results_broker` is using the ClickHouse [Kafka engine](https://clickhouse.com/docs/en/engines/table-engines/integrations/kafka) to fetch the results from Redpanda. The `osiris.results` table is used to store the results.
+The testbed consists in a Redpanda and ClickHouse instance. Required ClickHouse [tables](config/clickhouse/docker-entrypoint-initdb.d/init.sql) are created on startup. The `osiris.from_kafka` is using the ClickHouse [Kafka engine](https://clickhouse.com/docs/en/engines/table-engines/integrations/kafka) to fetch the results from Redpanda. The `osiris.results` table is used to store the results.
 
+As an example, Redpanda is configured with SASL authentication, and uses the default Osiris SASL credentials.
 
 ## Usage
 
@@ -13,10 +14,10 @@ The testbed consists in a Redpanda and ClickHouse instance. Required ClickHouse
 docker compose up -d --force-recreate --renew-anon-volumes
 ```
 
-* Run Osiris (from the root of the repository)
+* Run Osiris (from the root of the repository) with SASL authentication protocol
 
 ```sh
-cargo run
+cargo run -- --out-auth-protocol=SASL 2606:4700:4700::1111/128,1,32,1
 ```
 
 * Stop the testbed

testbed/README.md

@@ -1,8 +1,8 @@
 services:
   redpanda:
-    image: vectorized/redpanda:latest
-    command: redpanda start --check=false --overprovisioned --smp 1 --memory 500M
+    image: redpandadata/redpanda:latest
     volumes:
+    - ./config/redpanda/entrypoint.sh:/entrypoint.sh:ro
     - ./config/redpanda/redpanda.yml:/etc/redpanda/redpanda.yaml:ro
     ports:
     - "9092:9092"

testbed/compose.yml

@@ -1,5 +1,5 @@
-CREATE DATABASE osiris;
-CREATE TABLE osiris.results_broker
+CREATE DATABASE IF NOT EXISTS osiris;
+CREATE TABLE osiris.from_kafka
 (
     timestamp DateTime64,
     prober_id UInt16,
@@ -26,8 +26,8 @@ CREATE TABLE osiris.results_broker
 ENGINE = Kafka()
 SETTINGS
     kafka_broker_list = '10.0.0.100:9093',
-    kafka_topic_list = 'results',
-    kafka_group_name = 'clickhouse-results-group',
+    kafka_topic_list = 'osiris-results',
+    kafka_group_name = 'clickhouse-osiris-group',
     kafka_format = 'CSV';
 
 CREATE TABLE osiris.results
@@ -64,5 +64,5 @@ ORDER BY (
     probe_ttl
 );
 
-CREATE MATERIALIZED VIEW osiris.results_broker_mv TO osiris.results
-AS SELECT * FROM osiris.results_broker;
+CREATE MATERIALIZED VIEW osiris.from_kafka_mv TO osiris.results
+AS SELECT * FROM osiris.from_kafka;

testbed/config/clickhouse/docker-entrypoint-initdb.d/init.sql

@@ -0,0 +1,21 @@
+#!/bin/bash
+
+# Start Redpanda in the background
+set -m
+/usr/bin/rpk redpanda start --check=false --overprovisioned --smp 1 --memory 500M &
+
+# Wait for Redpanda to be ready
+sleep 3
+
+# Create `admin` superuser
+/usr/bin/rpk cluster config set superusers ['admin']
+/usr/bin/rpk security user create admin  -p 'admin' --mechanism=SCRAM-SHA-512
+
+# Enable SASL
+/usr/bin/rpk cluster config set enable_sasl true
+
+# Create `osiris` user and grant it access to the cluster and the `osiris-results` topic
+/usr/bin/rpk security user create osiris -p 'osiris' --mechanism SCRAM-SHA-512
+/usr/bin/rpk security acl create --allow-principal User:osiris --operation all --cluster -X user=admin -X pass='admin' -X sasl.mechanism=SCRAM-SHA-512
+/usr/bin/rpk security acl create --allow-principal User:osiris --operation all --topic osiris-results -X user=admin -X pass='admin' -X sasl.mechanism=SCRAM-SHA-512
+fg %1

testbed/config/redpanda/entrypoint.sh

@@ -7,11 +7,13 @@ redpanda:
         port: 33145
     kafka_api:
         - address: 0.0.0.0
-          name: internal
+          name: external
           port: 9092
+          authentication_method: sasl
         - address: 0.0.0.0
-          name: external
+          name: internal
           port: 9093
+          authentication_method: none
     admin:
         address: 0.0.0.0
         port: 9644
@@ -20,10 +22,10 @@ redpanda:
         port: 33145
     advertised_kafka_api:
         - address: localhost
-          name: internal
+          name: external
           port: 9092
         - address: 10.0.0.100
-          name: external
+          name: internal
           port: 9093
     developer_mode: true
     auto_create_topics_enabled: true