feat(config): Support SASL/OAUTHBEARER with an OIDC token endpoint

Cargo.toml

@@ -36,7 +36,7 @@ prost = "0.14"
 validator = { version = "0.20.0", features = ["derive"] }
 prost-types = "0.14"
 rand = "0.8.5"
-rdkafka = { version = "0.37.0", features = ["cmake-build", "ssl"] }
+rdkafka = { version = "0.37.0", features = ["cmake-build", "ssl", "curl"] }
 sentry_arroyo = { version = "=2.45.0", features = ["ssl"] }
 rmp-serde = "1.3"
 sentry = { version = "0.41.0", default-features = false, features = [

Dockerfile

@@ -4,7 +4,7 @@
 FROM ghcr.io/getsentry/image-mirror-library-rust:1.96.1-trixie AS build
 
 RUN apt-get update && apt-get upgrade -y
-RUN apt-get install -y cmake pkg-config libssl-dev librdkafka-dev protobuf-compiler
+RUN apt-get install -y cmake pkg-config libssl-dev libcurl4-openssl-dev librdkafka-dev protobuf-compiler
 
 RUN USER=root cargo new --bin taskbroker
 WORKDIR /taskbroker
@@ -40,7 +40,7 @@ FROM debian:trixie-slim
 RUN groupadd taskbroker --gid 1000 && useradd --gid taskbroker --uid 1000 taskbroker
 
 # Necessary for libssl bindings
-RUN apt-get update && apt-get upgrade -y && apt-get install -y libssl-dev libsqlite3-dev libsqlite3-0 openssl ca-certificates
+RUN apt-get update && apt-get upgrade -y && apt-get install -y libssl-dev libsqlite3-dev libsqlite3-0 openssl ca-certificates libcurl4
 
 EXPOSE 50051
 

README.md

@@ -93,6 +93,13 @@ kafka_clusters:
     # sasl_mechanism: scram-sha-256
     # sasl_username: ...
     # sasl_password: ...
+    # or OAUTHBEARER with tokens from an OIDC endpoint, eg. Google Managed
+    # Kafka's local auth server:
+    # sasl_mechanism: OAUTHBEARER
+    # sasl_oauthbearer_method: oidc
+    # sasl_oauthbearer_token_endpoint_url: http://localhost:14293
+    # sasl_oauthbearer_client_id: ...
+    # sasl_oauthbearer_client_secret: ...
 
 kafka_topics:
   taskworker:

src/config/kafka.rs

@@ -62,6 +62,14 @@ pub struct ClusterConfig {
     pub ssl_certificate_location: Option<String>,
     /// The location to the private key file
     pub ssl_key_location: Option<String>,
+    /// How librdkafka gets OAUTHBEARER tokens, eg. oidc
+    pub sasl_oauthbearer_method: Option<String>,
+    /// The OIDC token endpoint, eg. a local auth server
+    pub sasl_oauthbearer_token_endpoint_url: Option<String>,
+    /// The OIDC client id
+    pub sasl_oauthbearer_client_id: Option<String>,
+    /// The OIDC client secret
+    pub sasl_oauthbearer_client_secret: Option<String>,
 }
 
 impl ClusterConfig {
@@ -76,6 +84,10 @@ impl ClusterConfig {
             || self.ssl_ca_location.is_some()
             || self.ssl_certificate_location.is_some()
             || self.ssl_key_location.is_some()
+            || self.sasl_oauthbearer_method.is_some()
+            || self.sasl_oauthbearer_token_endpoint_url.is_some()
+            || self.sasl_oauthbearer_client_id.is_some()
+            || self.sasl_oauthbearer_client_secret.is_some()
     }
 
     /// Apply this cluster's `bootstrap.servers` and any configured sasl/ssl
@@ -105,5 +117,17 @@ impl ClusterConfig {
         if let Some(ref ssl_private_key_location) = self.ssl_key_location {
             config.set("ssl.key.location", ssl_private_key_location);
         }
+        if let Some(ref method) = self.sasl_oauthbearer_method {
+            config.set("sasl.oauthbearer.method", method);
+        }
+        if let Some(ref url) = self.sasl_oauthbearer_token_endpoint_url {
+            config.set("sasl.oauthbearer.token.endpoint.url", url);
+        }
+        if let Some(ref client_id) = self.sasl_oauthbearer_client_id {
+            config.set("sasl.oauthbearer.client.id", client_id);
+        }
+        if let Some(ref client_secret) = self.sasl_oauthbearer_client_secret {
+            config.set("sasl.oauthbearer.client.secret", client_secret);
+        }
     }
 }

src/config/mod.rs

@@ -349,6 +349,7 @@ impl Config {
 
         for cluster in config.kafka_clusters.values_mut() {
             redact_optional(&mut cluster.sasl_password);
+            redact_optional(&mut cluster.sasl_oauthbearer_client_secret);
         }
 
         // Redact strings
@@ -594,6 +595,10 @@ impl Config {
                         .kafka_ssl_certificate_location
                         .clone(),
                     ssl_key_location: self.deprecated.kafka_ssl_key_location.clone(),
+                    sasl_oauthbearer_method: None,
+                    sasl_oauthbearer_token_endpoint_url: None,
+                    sasl_oauthbearer_client_id: None,
+                    sasl_oauthbearer_client_secret: None,
                 },
             );
             assert!(
@@ -623,6 +628,10 @@ impl Config {
                         .kafka_deadletter_ssl_certificate_location
                         .clone(),
                     ssl_key_location: self.deprecated.kafka_deadletter_ssl_key_location.clone(),
+                    sasl_oauthbearer_method: None,
+                    sasl_oauthbearer_token_endpoint_url: None,
+                    sasl_oauthbearer_client_id: None,
+                    sasl_oauthbearer_client_secret: None,
                 },
             );
             assert!(
@@ -1723,6 +1732,10 @@ kafka_clusters:
                     ssl_ca_location: None,
                     ssl_certificate_location: None,
                     ssl_key_location: None,
+                    sasl_oauthbearer_method: None,
+                    sasl_oauthbearer_token_endpoint_url: None,
+                    sasl_oauthbearer_client_id: None,
+                    sasl_oauthbearer_client_secret: None,
                 })
             );
             assert!(!clusters.contains_key("default"));