From b97dacdc8b0f0e19ed70a3a8d9f67fb97da4aa3e Mon Sep 17 00:00:00 2001 From: Enoch Tang Date: Sat, 24 Oct 2026 19:50:42 -0400 Subject: [PATCH] feat(config): Support SASL/OAUTHBEARER with an OIDC token endpoint Adds sasl_oauthbearer_method, sasl_oauthbearer_token_endpoint_url, sasl_oauthbearer_client_id and sasl_oauthbearer_client_secret to kafka_clusters and passes them to librdkafka. This lets taskbroker log in to Google Managed Kafka through Google's local auth server, which hands out Workload Identity tokens, so no Kafka password is stored. librdkafka needs libcurl to fetch tokens over HTTP, so this turns on rdkafka's curl feature and adds libcurl to the image. Only the kafka_clusters format gets the new settings, not the deprecated flat kafka_* fields. --- Cargo.lock | 16 ++++++++++++++++ Cargo.toml | 2 +- Dockerfile | 4 ++-- README.md | 7 +++++++ src/config/kafka.rs | 24 ++++++++++++++++++++++++ src/config/mod.rs | 13 +++++++++++++ 6 files changed, 63 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2a258356..aacfda13 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -589,6 +589,21 @@ dependencies = [ "typenum", ] +[[package]] +name = "curl-sys" +version = "0.4.91+curl-8.22.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd39579123e45bf762716bb893d43886ed1cabe6289455be29a4f1d4e37afb06" +dependencies = [ + "cc", + "libc", + "libz-sys", + "openssl-sys", + "pkg-config", + "vcpkg", + "windows-sys 0.61.2", +] + [[package]] name = "darling" version = "0.20.11" @@ -2525,6 +2540,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e234cf318915c1059d4921ef7f75616b5219b10b46e9f3a511a15eb4b56a3f77" dependencies = [ "cmake", + "curl-sys", "libc", "libz-sys", "num_enum", diff --git a/Cargo.toml b/Cargo.toml index 0470aef2..a995db28 100644 --- a/Cargo.toml +++ b/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 = [ diff --git a/Dockerfile b/Dockerfile index d6ebe91c..d704dc38 100644 --- a/Dockerfile +++ b/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 diff --git a/README.md b/README.md index f8ae89c1..9f50b119 100644 --- a/README.md +++ b/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: diff --git a/src/config/kafka.rs b/src/config/kafka.rs index 522bc207..fbddd45d 100644 --- a/src/config/kafka.rs +++ b/src/config/kafka.rs @@ -62,6 +62,14 @@ pub struct ClusterConfig { pub ssl_certificate_location: Option, /// The location to the private key file pub ssl_key_location: Option, + /// How librdkafka gets OAUTHBEARER tokens, eg. oidc + pub sasl_oauthbearer_method: Option, + /// The OIDC token endpoint, eg. a local auth server + pub sasl_oauthbearer_token_endpoint_url: Option, + /// The OIDC client id + pub sasl_oauthbearer_client_id: Option, + /// The OIDC client secret + pub sasl_oauthbearer_client_secret: Option, } 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); + } } } diff --git a/src/config/mod.rs b/src/config/mod.rs index 3c5289f3..a6fd85ed 100644 --- a/src/config/mod.rs +++ b/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"));