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"));