From d6165629b2e7c59a63c6029b047f5981a1aa1a23 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Mon, 6 Jul 2026 13:49:42 +0200 Subject: [PATCH 01/10] feat(kafka source, kafka sink): support AWS MSK IAM auth via SASL/OAUTHBEARER MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit librdkafka has no native AWS_MSK_IAM SASL mechanism (that lives only in the JVM client), so the portable path to authenticate against MSK with IAM is SASL/OAUTHBEARER, where the OAuth token is a SigV4-presigned kafka-cluster:Connect request, base64url-encoded. This ports the official aws-msk-iam-sasl-signer recipe. Add a sasl.aws_msk_iam config block (region + standard AwsAuthentication) to the kafka source and sink. When enabled, the mechanism is forced to OAUTHBEARER and KafkaStatisticsContext::generate_oauth_token mints/refreshes the token, reusing the existing AWS credential providers (static/profile/assume-role/IMDS). Gated behind the aws-core feature. Token generation correctness is proven offline against an independent SigV4 oracle (fixed inputs -> known-answer signature and full presigned URL). Closes #1516 Signed-off-by: Gaál György --- Cargo.toml | 1 + changelog.d/kafka_msk_iam_auth.feature.md | 3 + src/kafka.rs | 188 +++++++++++++- src/kafka/msk_iam.rs | 291 ++++++++++++++++++++++ src/sinks/kafka/config.rs | 17 +- src/sinks/kafka/sink.rs | 31 ++- src/sinks/kafka/tests.rs | 57 ++++- src/sources/kafka.rs | 50 +++- 8 files changed, 602 insertions(+), 36 deletions(-) create mode 100644 changelog.d/kafka_msk_iam_auth.feature.md create mode 100644 src/kafka/msk_iam.rs diff --git a/Cargo.toml b/Cargo.toml index 391ab9c4394ba..40b67d31ff659 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -650,6 +650,7 @@ aws-core = [ "dep:aws-smithy-runtime", "dep:aws-smithy-runtime-api", "dep:aws-sdk-sts", + "dep:base64", ] # Anything that requires Protocol Buffers. diff --git a/changelog.d/kafka_msk_iam_auth.feature.md b/changelog.d/kafka_msk_iam_auth.feature.md new file mode 100644 index 0000000000000..bb5c831c0c548 --- /dev/null +++ b/changelog.d/kafka_msk_iam_auth.feature.md @@ -0,0 +1,3 @@ +The `kafka` source and sink now support AWS MSK IAM authentication via SASL/OAUTHBEARER. Configure it under `sasl.aws_msk_iam` with a region and the standard AWS authentication options (static keys, profile, assumed role, or IMDS). The OAuth token is a SigV4-presigned `kafka-cluster:Connect` request, generated and refreshed automatically, so no static SCRAM credentials are needed. Requires TLS (MSK IAM listens on the SASL_SSL port, `9098`). + +authors: gecube diff --git a/src/kafka.rs b/src/kafka.rs index 0c9315288e313..668a10385c2d8 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -11,6 +11,23 @@ use crate::{ tls::{PEM_START_MARKER, TlsEnableableConfig}, }; +#[cfg(feature = "aws-core")] +mod msk_iam; + +#[cfg(feature = "aws-core")] +use aws_credential_types::provider::SharedCredentialsProvider; +#[cfg(feature = "aws-core")] +use aws_types::region::Region; +#[cfg(feature = "aws-core")] +use rdkafka::client::OAuthToken; +#[cfg(feature = "aws-core")] +use tokio::runtime::Handle; +#[cfg(feature = "aws-core")] +use crate::{ + aws::{AwsAuthentication, RegionOrEndpoint}, + config::ProxyConfig, +}; + #[derive(Debug, Snafu)] enum KafkaError { #[snafu(display("invalid path: {:?}", path))] @@ -80,6 +97,33 @@ pub struct KafkaSaslConfig { #[configurable(metadata(docs::examples = "SCRAM-SHA-256"))] #[configurable(metadata(docs::examples = "SCRAM-SHA-512"))] pub(crate) mechanism: Option, + + /// AWS MSK IAM authentication using SASL/OAUTHBEARER. + /// + /// When enabled, `mechanism` is forced to `OAUTHBEARER` and the token is generated by + /// SigV4-presigning the `kafka-cluster:Connect` action against the configured region. Mutually + /// exclusive with `username`/`password`. Requires TLS (MSK IAM listens on the SASL_SSL port, + /// `9098`). + #[cfg(feature = "aws-core")] + #[configurable(derived)] + pub(crate) aws_msk_iam: Option, +} + +/// Configuration for AWS MSK IAM authentication (SASL/OAUTHBEARER). +#[cfg(feature = "aws-core")] +#[configurable_component] +#[derive(Clone, Debug, Default)] +pub struct AwsMskIamConfig { + /// Enables MSK IAM authentication. + pub(crate) enabled: bool, + + #[configurable(derived)] + #[serde(flatten, default)] + pub(crate) region: RegionOrEndpoint, + + #[configurable(derived)] + #[serde(default)] + pub(crate) auth: AwsAuthentication, } impl KafkaAuthConfig { @@ -97,14 +141,30 @@ impl KafkaAuthConfig { if sasl_enabled { let sasl = self.sasl.as_ref().unwrap(); - if let Some(username) = &sasl.username { - client.set("sasl.username", username.as_str()); - } - if let Some(password) = &sasl.password { - client.set("sasl.password", password.inner()); - } - if let Some(mechanism) = &sasl.mechanism { - client.set("sasl.mechanism", mechanism); + + #[cfg(feature = "aws-core")] + let msk_iam_enabled = sasl + .aws_msk_iam + .as_ref() + .map(|c| c.enabled) + .unwrap_or(false); + #[cfg(not(feature = "aws-core"))] + let msk_iam_enabled = false; + + if msk_iam_enabled { + // OAUTHBEARER: the token itself is produced by + // `KafkaStatisticsContext::generate_oauth_token`. No username/password. + client.set("sasl.mechanism", "OAUTHBEARER"); + } else { + if let Some(username) = &sasl.username { + client.set("sasl.username", username.as_str()); + } + if let Some(password) = &sasl.password { + client.set("sasl.password", password.inner()); + } + if let Some(mechanism) = &sasl.mechanism { + client.set("sasl.mechanism", mechanism); + } } } @@ -159,6 +219,39 @@ impl KafkaAuthConfig { Ok(()) } + + /// Resolve the AWS credential material needed by the OAUTHBEARER token callback, if (and only + /// if) MSK IAM auth is enabled. Async because building the credentials provider (IMDS / STS / + /// SSO) is async. Call once at build time and move the result into `KafkaStatisticsContext`. + #[cfg(feature = "aws-core")] + pub(crate) async fn msk_iam_credentials( + &self, + proxy: &ProxyConfig, + ) -> crate::Result> { + let Some(cfg) = self.sasl.as_ref().and_then(|s| s.aws_msk_iam.as_ref()) else { + return Ok(None); + }; + if !cfg.enabled { + return Ok(None); + } + let region = cfg.region.region().ok_or( + "`sasl.aws_msk_iam` requires a region (set `sasl.aws_msk_iam.region`)", + )?; + let provider = cfg + .auth + .credentials_provider(region.clone(), proxy, None) + .await?; + Ok(Some(MskIamCredentials { region, provider })) + } +} + +/// AWS credential material captured at build time and moved into the rdkafka client context so the +/// (synchronous) OAUTHBEARER refresh callback can mint fresh tokens. +#[cfg(feature = "aws-core")] +#[derive(Clone)] +pub(crate) struct MskIamCredentials { + pub(crate) region: Region, + pub(crate) provider: SharedCredentialsProvider, } fn pathbuf_to_string(path: &Path) -> crate::Result<&str> { @@ -169,9 +262,46 @@ fn pathbuf_to_string(path: &Path) -> crate::Result<&str> { pub(crate) struct KafkaStatisticsContext { pub(crate) expose_lag_metrics: bool, pub span: Span, + /// Present only when MSK IAM (SASL/OAUTHBEARER) auth is enabled. + #[cfg(feature = "aws-core")] + pub(crate) msk_iam: Option, + /// Handle to the tokio runtime, captured at construction. The OAUTHBEARER callback runs on + /// librdkafka's own (non-tokio) thread, so it bridges to async credential resolution via + /// `Handle::block_on`. `Option` because construction may happen outside a runtime (e.g. tests). + #[cfg(feature = "aws-core")] + pub(crate) runtime: Option, +} + +impl KafkaStatisticsContext { + #[cfg(feature = "aws-core")] + pub(crate) fn new( + expose_lag_metrics: bool, + span: Span, + msk_iam: Option, + ) -> Self { + Self { + expose_lag_metrics, + span, + msk_iam, + runtime: Handle::try_current().ok(), + } + } + + #[cfg(not(feature = "aws-core"))] + pub(crate) fn new(expose_lag_metrics: bool, span: Span) -> Self { + Self { + expose_lag_metrics, + span, + } + } } impl ClientContext for KafkaStatisticsContext { + // Required for `generate_oauth_token` to be invoked by librdkafka. Harmless for non-OAUTHBEARER + // mechanisms (SCRAM/PLAIN), where the callback is simply never called. + #[cfg(feature = "aws-core")] + const ENABLE_REFRESH_OAUTH_TOKEN: bool = true; + fn stats(&self, statistics: Statistics) { // This callback get executed on a separate thread within the rdkafka library, so we need // to propagate the span here to attach the component tags to the emitted events. @@ -181,6 +311,48 @@ impl ClientContext for KafkaStatisticsContext { expose_lag_metrics: self.expose_lag_metrics, }); } + + #[cfg(feature = "aws-core")] + fn generate_oauth_token( + &self, + _oauthbearer_config: Option<&str>, + ) -> Result> { + let creds = self + .msk_iam + .as_ref() + .ok_or("generate_oauth_token called but MSK IAM is not configured")?; + let runtime = self + .runtime + .as_ref() + .ok_or("no tokio runtime handle available to mint the MSK IAM token")?; + + // The MSK signer signs a synthetic `kafka.{region}.amazonaws.com` host, so region + + // credentials are all the callback needs. block_on is safe: this runs on librdkafka's own + // (non-tokio) thread, so there is no runtime nesting. + let token = + runtime.block_on(msk_iam::generate_auth_token(&creds.region, &creds.provider))?; + + Ok(OAuthToken { + token: token.token, + // MSK ignores the principal name, but librdkafka requires it to be non-empty. + principal_name: "vector".to_string(), + lifetime_ms: token.lifetime_ms, + }) + } } impl ConsumerContext for KafkaStatisticsContext {} + +// Enables `KafkaStatisticsContext` to back a `BaseProducer` directly (used by the sink healthcheck), +// so the OAUTHBEARER token callback is available there too. The sink's `FutureProducer` wraps this +// context separately and is unaffected. +impl rdkafka::producer::ProducerContext for KafkaStatisticsContext { + type DeliveryOpaque = (); + + fn delivery( + &self, + _delivery_result: &rdkafka::producer::DeliveryResult<'_>, + _delivery_opaque: Self::DeliveryOpaque, + ) { + } +} diff --git a/src/kafka/msk_iam.rs b/src/kafka/msk_iam.rs new file mode 100644 index 0000000000000..4edae7c86952d --- /dev/null +++ b/src/kafka/msk_iam.rs @@ -0,0 +1,291 @@ +//! AWS MSK IAM authentication for the `kafka` source and sink (SASL/OAUTHBEARER). +//! +//! librdkafka has no native `AWS_MSK_IAM` SASL mechanism (that lives only in the JVM client). The +//! portable path is SASL/OAUTHBEARER, where the OAuth token is a SigV4-presigned URL for the +//! `kafka-cluster:Connect` action, base64url-encoded. This is a port of the official +//! `aws-msk-iam-sasl-signer-go` `constructAuthToken`; correctness is proven offline in the tests +//! below against an independent SigV4 oracle (fixed inputs -> known-answer signature). + +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use aws_credential_types::Credentials; +use aws_credential_types::provider::{ProvideCredentials, SharedCredentialsProvider}; +use aws_sigv4::http_request::{ + SignableBody, SignableRequest, SignatureLocation, SigningSettings, sign, +}; +use aws_sigv4::sign::v4; +use aws_smithy_runtime_api::client::identity::Identity; +use aws_types::region::Region; +use base64::Engine as _; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use snafu::{ResultExt, Snafu}; + +const EXPIRY_SECONDS: u64 = 900; +const SIGNING_NAME: &str = "kafka-cluster"; +const ACTION: &str = "kafka-cluster:Connect"; +const USER_AGENT: &str = concat!("vector-msk-iam-signer/", env!("CARGO_PKG_VERSION")); + +type BoxError = Box; + +#[derive(Debug, Snafu)] +pub enum MskIamError { + #[snafu(display("failed to load AWS credentials for MSK IAM auth: {source}"))] + Credentials { + source: aws_credential_types::provider::error::CredentialsError, + }, + #[snafu(display("failed to build SigV4 signing params: {source}"))] + SigningParams { source: BoxError }, + #[snafu(display("failed to presign MSK IAM token: {source}"))] + Signing { source: BoxError }, +} + +/// Token librdkafka hands to the broker, plus its absolute expiry (unix-epoch ms). +#[derive(Debug, Clone)] +pub struct MskAuthToken { + pub token: String, + pub lifetime_ms: i64, +} + +/// Generate a fresh MSK IAM SASL/OAUTHBEARER token for `region`. +pub async fn generate_auth_token( + region: &Region, + credentials_provider: &SharedCredentialsProvider, +) -> Result { + let credentials = credentials_provider + .provide_credentials() + .await + .context(CredentialsSnafu)?; + let signing_time = SystemTime::now(); + build_token(region, &credentials, signing_time) +} + +/// Deterministic core: given credentials + a fixed signing time, produce the token. Split out so +/// tests can pin the time and assert a known-answer signature. +fn build_token( + region: &Region, + credentials: &Credentials, + signing_time: SystemTime, +) -> Result { + let signed_url = presign(region, credentials, signing_time)?; + // `User-Agent` is appended *after* signing (unsigned), matching the AWS signers. + let signed_url = format!("{signed_url}&User-Agent={USER_AGENT}"); + let token = URL_SAFE_NO_PAD.encode(signed_url.as_bytes()); + + let signing_ms = signing_time + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64; + Ok(MskAuthToken { + token, + lifetime_ms: signing_ms + (EXPIRY_SECONDS as i64) * 1000, + }) +} + +/// Build the SigV4-presigned `https://kafka.{region}.amazonaws.com/?Action=...` URL. +fn presign( + region: &Region, + credentials: &Credentials, + time: SystemTime, +) -> Result { + let host = format!("kafka.{}.amazonaws.com", region.as_ref()); + let identity: Identity = credentials.clone().into(); + + let mut settings = SigningSettings::default(); + settings.signature_location = SignatureLocation::QueryParams; + settings.expires_in = Some(Duration::from_secs(EXPIRY_SECONDS)); + + let signing_params = v4::SigningParams::builder() + .identity(&identity) + .region(region.as_ref()) + .name(SIGNING_NAME) + .time(time) + .settings(settings) + .build() + .map_err(|e| MskIamError::SigningParams { source: Box::new(e) })?; + + let url = format!("https://{host}/?Action={}", rfc3986(ACTION)); + let headers = [("host", host.as_str())]; + let signable = SignableRequest::new("GET", &url, headers.into_iter(), SignableBody::Bytes(&[])) + .map_err(|e| MskIamError::Signing { source: Box::new(e) })?; + + let output = sign(signable, &signing_params.into()) + .map_err(|e| MskIamError::Signing { source: Box::new(e) })?; + let (instructions, _signature) = output.into_parts(); + + // `params()` yields DECODED (name, value) pairs. They must be re-encoded with the exact RFC3986 + // rules AWS used when signing the canonical request, or the emitted URL won't match its own + // signature and the broker will reject the token (e.g. `/` in X-Amz-Credential -> `%2F`). + let mut final_url = url.clone(); + for (name, value) in instructions.params() { + final_url.push('&'); + final_url.push_str(&rfc3986(name)); + final_url.push('='); + final_url.push_str(&rfc3986(value)); + } + Ok(final_url) +} + +/// RFC3986 percent-encoding matching AWS SigV4 canonicalization (unreserved = A-Za-z0-9-_.~). +fn rfc3986(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + for &b in s.as_bytes() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + out.push(b as char) + } + _ => out.push_str(&format!("%{b:02X}")), + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + // Fixed inputs shared with oracle/sigv4_oracle.py. 1_700_000_000 == 2023-11-14T22:13:20Z. + const ACCESS_KEY: &str = "AKIDEXAMPLE"; + const SECRET_KEY: &str = "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY"; + const SESSION_TOKEN: &str = "IQoJb3Jp"; + const FIXED_EPOCH_SECS: u64 = 1_700_000_000; + + // Golden signatures produced by the independent Python oracle. + const GOLDEN_SIG_WITH_TOKEN: &str = + "4c02acd7700d674cecd2b27f6a906be056df725db5c7571700fd54a354a31904"; + const GOLDEN_SIG_NO_TOKEN: &str = + "12f697bba1fe524b7e9f370e2c56075ea612d9d4d2c7f16c402ffc72520fce5e"; + + fn region() -> Region { + Region::new("ap-southeast-1") + } + + fn fixed_time() -> SystemTime { + UNIX_EPOCH + Duration::from_secs(FIXED_EPOCH_SECS) + } + + fn query_param<'a>(url: &'a str, key: &str) -> Option<&'a str> { + let q = url.split_once('?')?.1; + q.split('&').find_map(|kv| { + let (k, v) = kv.split_once('=')?; + (k == key).then_some(v) + }) + } + + /// Split a URL into (base, sorted [(key,value)]) so two URLs can be compared for identical + /// content regardless of query-parameter ordering (MSK re-canonicalizes, so order is irrelevant). + fn normalize(url: &str) -> (String, Vec<(String, String)>) { + let (base, query) = url.split_once('?').expect("url has query"); + let mut params: Vec<(String, String)> = query + .split('&') + .map(|kv| { + let (k, v) = kv.split_once('=').expect("kv"); + (k.to_string(), v.to_string()) + }) + .collect(); + params.sort(); + (base.to_string(), params) + } + + // Full presigned URLs from the independent oracle (oracle/sigv4_oracle.py output). + const ORACLE_URL_WITH_TOKEN: &str = "https://kafka.ap-southeast-1.amazonaws.com/?Action=kafka-cluster%3AConnect&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIDEXAMPLE%2F20231114%2Fap-southeast-1%2Fkafka-cluster%2Faws4_request&X-Amz-Date=20231114T221320Z&X-Amz-Expires=900&X-Amz-Security-Token=IQoJb3Jp&X-Amz-SignedHeaders=host&X-Amz-Signature=4c02acd7700d674cecd2b27f6a906be056df725db5c7571700fd54a354a31904"; + + /// STRONGEST PROOF: the *entire* presigned URL (every param, every encoded value, the signature) + /// matches the independent oracle — not just the signature. Order-independent because MSK + /// re-canonicalizes the query on its side. + #[test] + fn full_presigned_url_matches_oracle() { + let creds = + Credentials::from_keys(ACCESS_KEY, SECRET_KEY, Some(SESSION_TOKEN.to_string())); + let url = presign(®ion(), &creds, fixed_time()).expect("presign"); + assert_eq!(normalize(&url), normalize(ORACLE_URL_WITH_TOKEN), "url={url}"); + } + + /// KNOWN-ANSWER: the aws-sigv4 signature must equal the independent oracle's, byte-for-byte, + /// with a session token present (X-Amz-Security-Token in the signed canonical query). + #[test] + fn signature_matches_oracle_with_session_token() { + let creds = + Credentials::from_keys(ACCESS_KEY, SECRET_KEY, Some(SESSION_TOKEN.to_string())); + let url = presign(®ion(), &creds, fixed_time()).expect("presign"); + + assert_eq!( + query_param(&url, "X-Amz-Signature"), + Some(GOLDEN_SIG_WITH_TOKEN), + "aws-sigv4 signature diverged from the independent SigV4 oracle\nurl={url}" + ); + assert_eq!(query_param(&url, "X-Amz-Security-Token"), Some("IQoJb3Jp")); + } + + /// KNOWN-ANSWER: same, without a session token (static long-term credentials). + #[test] + fn signature_matches_oracle_without_session_token() { + let creds = Credentials::from_keys(ACCESS_KEY, SECRET_KEY, None); + let url = presign(®ion(), &creds, fixed_time()).expect("presign"); + + assert_eq!( + query_param(&url, "X-Amz-Signature"), + Some(GOLDEN_SIG_NO_TOKEN), + "aws-sigv4 signature diverged from the independent SigV4 oracle\nurl={url}" + ); + assert_eq!(query_param(&url, "X-Amz-Security-Token"), None); + } + + /// Structural invariants the MSK broker requires of the presigned URL. + #[test] + fn presigned_url_has_required_msk_shape() { + let creds = Credentials::from_keys(ACCESS_KEY, SECRET_KEY, None); + let url = presign(®ion(), &creds, fixed_time()).expect("presign"); + + assert!(url.starts_with("https://kafka.ap-southeast-1.amazonaws.com/?")); + assert_eq!(query_param(&url, "Action"), Some("kafka-cluster%3AConnect")); + assert_eq!(query_param(&url, "X-Amz-Algorithm"), Some("AWS4-HMAC-SHA256")); + assert_eq!(query_param(&url, "X-Amz-Expires"), Some("900")); + assert_eq!(query_param(&url, "X-Amz-SignedHeaders"), Some("host")); + assert_eq!( + query_param(&url, "X-Amz-Credential"), + Some("AKIDEXAMPLE%2F20231114%2Fap-southeast-1%2Fkafka-cluster%2Faws4_request") + ); + } + + /// The final token is base64url-nopad of the presigned URL (+ appended User-Agent). + #[test] + fn token_is_base64url_nopad_of_presigned_url() { + let creds = Credentials::from_keys(ACCESS_KEY, SECRET_KEY, None); + let out = build_token(®ion(), &creds, fixed_time()).expect("token"); + + assert!(!out.token.contains('='), "token must be unpadded base64url"); + let decoded = URL_SAFE_NO_PAD.decode(out.token.as_bytes()).expect("decode"); + let url = String::from_utf8(decoded).expect("utf8"); + assert!(url.starts_with("https://kafka.ap-southeast-1.amazonaws.com/?")); + assert!(url.ends_with(&format!("&User-Agent={USER_AGENT}"))); + assert!(url.contains(GOLDEN_SIG_NO_TOKEN)); + + // Absolute expiry = signing time + 900s, in unix-epoch ms. + assert_eq!(out.lifetime_ms, (FIXED_EPOCH_SECS as i64) * 1000 + 900_000); + } + + /// End-to-end through the public async entrypoint with a static credentials provider. + #[tokio::test] + async fn generate_auth_token_via_provider() { + let provider = SharedCredentialsProvider::new(Credentials::from_keys( + ACCESS_KEY, + SECRET_KEY, + Some(SESSION_TOKEN.to_string()), + )); + let out = generate_auth_token(®ion(), &provider) + .await + .expect("generate"); + + let decoded = URL_SAFE_NO_PAD.decode(out.token.as_bytes()).expect("decode"); + let url = String::from_utf8(decoded).expect("utf8"); + assert!(url.contains("Action=kafka-cluster%3AConnect")); + assert!(url.contains("X-Amz-Signature=")); + // Uses wall-clock now(): lifetime must be ~900s in the future, so comfortably positive. + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as i64; + assert!(out.lifetime_ms > now_ms); + assert!(out.lifetime_ms <= now_ms + 900_000 + 5_000); + } +} diff --git a/src/sinks/kafka/config.rs b/src/sinks/kafka/config.rs index b1e00a3e0c32c..f5921d1e81feb 100644 --- a/src/sinks/kafka/config.rs +++ b/src/sinks/kafka/config.rs @@ -280,8 +280,21 @@ impl GenerateConfig for KafkaSinkConfig { #[typetag::serde(name = "kafka")] impl SinkConfig for KafkaSinkConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { - let sink = KafkaSink::new(self.clone())?; - let hc = healthcheck(self.clone(), cx.healthcheck.clone()).boxed(); + #[cfg(feature = "aws-core")] + let msk_iam = self.auth.msk_iam_credentials(&cx.proxy).await?; + + let sink = KafkaSink::new( + self.clone(), + #[cfg(feature = "aws-core")] + msk_iam.clone(), + )?; + let hc = healthcheck( + self.clone(), + cx.healthcheck.clone(), + #[cfg(feature = "aws-core")] + msk_iam, + ) + .boxed(); Ok((VectorSink::from_event_streamsink(sink), hc)) } diff --git a/src/sinks/kafka/sink.rs b/src/sinks/kafka/sink.rs index 9b930fd4abecd..e6b1eb91892ac 100644 --- a/src/sinks/kafka/sink.rs +++ b/src/sinks/kafka/sink.rs @@ -38,20 +38,29 @@ pub struct KafkaSink { pub(crate) fn create_producer( client_config: ClientConfig, + #[cfg(feature = "aws-core")] msk_iam: Option, ) -> crate::Result> { + #[cfg(feature = "aws-core")] + let context = KafkaStatisticsContext::new(false, Span::current(), msk_iam); + #[cfg(not(feature = "aws-core"))] + let context = KafkaStatisticsContext::new(false, Span::current()); let producer = client_config - .create_with_context(KafkaStatisticsContext { - expose_lag_metrics: false, - span: Span::current(), - }) + .create_with_context(context) .context(KafkaCreateFailedSnafu)?; Ok(producer) } impl KafkaSink { - pub(crate) fn new(config: KafkaSinkConfig) -> crate::Result { + pub(crate) fn new( + config: KafkaSinkConfig, + #[cfg(feature = "aws-core")] msk_iam: Option, + ) -> crate::Result { let producer_config = config.to_rdkafka()?; - let producer = create_producer(producer_config)?; + let producer = create_producer( + producer_config, + #[cfg(feature = "aws-core")] + msk_iam, + )?; let transformer = config.encoding.transformer(); let serializer = config.encoding.build()?; let encoder = Encoder::<()>::new(serializer); @@ -115,6 +124,7 @@ impl KafkaSink { pub(crate) async fn healthcheck( config: KafkaSinkConfig, healthcheck_options: SinkHealthcheckOptions, + #[cfg(feature = "aws-core")] msk_iam: Option, ) -> crate::Result<()> { trace!("Healthcheck started."); let client_config = config.to_rdkafka().unwrap(); @@ -132,8 +142,15 @@ pub(crate) async fn healthcheck( }, }; + // Build the client context in the async context (so the tokio runtime handle is captured for + // the OAUTHBEARER token callback), then move it into the blocking task. + #[cfg(feature = "aws-core")] + let context = KafkaStatisticsContext::new(false, Span::current(), msk_iam); + #[cfg(not(feature = "aws-core"))] + let context = KafkaStatisticsContext::new(false, Span::current()); + tokio::task::spawn_blocking(move || { - let producer: BaseProducer = client_config.create().unwrap(); + let producer: BaseProducer<_> = client_config.create_with_context(context).unwrap(); let topic = topic.as_deref(); producer diff --git a/src/sinks/kafka/tests.rs b/src/sinks/kafka/tests.rs index 845b17e44061b..ab72f3402b630 100644 --- a/src/sinks/kafka/tests.rs +++ b/src/sinks/kafka/tests.rs @@ -65,9 +65,14 @@ mod integration_test { headers_key: None, acknowledgements: Default::default(), }; - self::sink::healthcheck(config, Default::default()) - .await - .unwrap(); + self::sink::healthcheck( + config, + Default::default(), + #[cfg(feature = "aws-core")] + None, + ) + .await + .unwrap(); } #[tokio::test] @@ -93,9 +98,14 @@ mod integration_test { headers_key: None, acknowledgements: Default::default(), }; - self::sink::healthcheck(config, Default::default()) - .await - .unwrap(); + self::sink::healthcheck( + config, + Default::default(), + #[cfg(feature = "aws-core")] + None, + ) + .await + .unwrap(); } #[tokio::test] @@ -197,8 +207,18 @@ mod integration_test { acknowledgements: Default::default(), }; config.clone().to_rdkafka()?; - self::sink::healthcheck(config.clone(), Default::default()).await?; - KafkaSink::new(config) + self::sink::healthcheck( + config.clone(), + Default::default(), + #[cfg(feature = "aws-core")] + None, + ) + .await?; + KafkaSink::new( + config, + #[cfg(feature = "aws-core")] + None, + ) } #[tokio::test] @@ -363,7 +383,12 @@ mod integration_test { events.push(Event::Trace(trace.with_batch_notifier(&batch))); } - let sink = KafkaSink::new(config).unwrap(); + let sink = KafkaSink::new( + config, + #[cfg(feature = "aws-core")] + None, + ) + .unwrap(); let sink = VectorSink::from_event_streamsink(sink); let stream = map_event_batch_stream(stream::iter(events), Some(batch)); sink.run(stream).await.unwrap(); @@ -497,7 +522,12 @@ mod integration_test { if test_telemetry_tags { assert_data_volume_sink_compliance(&DATA_VOLUME_SINK_TAGS, async move { - let sink = KafkaSink::new(config).unwrap(); + let sink = KafkaSink::new( + config, + #[cfg(feature = "aws-core")] + None, + ) + .unwrap(); let sink = VectorSink::from_event_streamsink(sink); sink.run(input_events).await }) @@ -505,7 +535,12 @@ mod integration_test { .expect("Running sink failed"); } else { assert_sink_compliance(&SINK_TAGS, async move { - let sink = KafkaSink::new(config).unwrap(); + let sink = KafkaSink::new( + config, + #[cfg(feature = "aws-core")] + None, + ) + .unwrap(); let sink = VectorSink::from_event_streamsink(sink); sink.run(input_events).await }) diff --git a/src/sources/kafka.rs b/src/sources/kafka.rs index 1a14680bc4ee6..924927954d87c 100644 --- a/src/sources/kafka.rs +++ b/src/sources/kafka.rs @@ -342,7 +342,15 @@ impl SourceConfig for KafkaSourceConfig { ); } - let (consumer, callback_rx) = create_consumer(self, acknowledgements)?; + #[cfg(feature = "aws-core")] + let msk_iam = self.auth.msk_iam_credentials(&cx.proxy).await?; + + let (consumer, callback_rx) = create_consumer( + self, + acknowledgements, + #[cfg(feature = "aws-core")] + msk_iam, + )?; Ok(Box::pin(kafka_source( self.clone(), @@ -1188,6 +1196,7 @@ impl<'a> From> for FinalizerEntry { fn create_consumer( config: &KafkaSourceConfig, acknowledgements: bool, + #[cfg(feature = "aws-core")] msk_iam: Option, ) -> crate::Result<( StreamConsumer, UnboundedReceiver, @@ -1234,6 +1243,8 @@ fn create_consumer( acknowledgements, callbacks, Span::current(), + #[cfg(feature = "aws-core")] + msk_iam, )) .context(CreateSnafu)?; @@ -1274,12 +1285,13 @@ impl KafkaSourceContext { acknowledgements: bool, callbacks: UnboundedSender, span: Span, + #[cfg(feature = "aws-core")] msk_iam: Option, ) -> Self { Self { - stats: kafka::KafkaStatisticsContext { - expose_lag_metrics, - span, - }, + #[cfg(feature = "aws-core")] + stats: kafka::KafkaStatisticsContext::new(expose_lag_metrics, span, msk_iam), + #[cfg(not(feature = "aws-core"))] + stats: kafka::KafkaStatisticsContext::new(expose_lag_metrics, span), acknowledgements, consumer: OnceLock::default(), callbacks, @@ -1522,7 +1534,15 @@ mod test { #[tokio::test] async fn consumer_create_ok() { let config = make_config("topic", "group", LogNamespace::Legacy, None); - assert!(create_consumer(&config, true).is_ok()); + assert!( + create_consumer( + &config, + true, + #[cfg(feature = "aws-core")] + None + ) + .is_ok() + ); } #[tokio::test] @@ -1531,7 +1551,15 @@ mod test { auto_offset_reset: "incorrect-auto-offset-reset".to_string(), ..make_config("topic", "group", LogNamespace::Legacy, None) }; - assert!(create_consumer(&config, true).is_err()); + assert!( + create_consumer( + &config, + true, + #[cfg(feature = "aws-core")] + None + ) + .is_err() + ); } } @@ -1816,7 +1844,13 @@ mod integration_test { .build() .unwrap(); - let (consumer, callback_rx) = create_consumer(&config, acknowledgements).unwrap(); + let (consumer, callback_rx) = create_consumer( + &config, + acknowledgements, + #[cfg(feature = "aws-core")] + None, + ) + .unwrap(); tokio::spawn(kafka_source( config, From 27927206e23db81f638e1da910e6723e4ff7d64f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Mon, 6 Jul 2026 15:53:05 +0200 Subject: [PATCH 02/10] fix(kafka): honor MSK IAM on source context and enable-by-flag; add auth matrix tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses review feedback: - Source: the consumer's rdkafka context is KafkaSourceContext (not the inner KafkaStatisticsContext), so it now enables ENABLE_REFRESH_OAUTH_TOKEN and forwards generate_oauth_token to the inner context. Without this, MSK IAM sources were configured with sasl.mechanism=OAUTHBEARER but had no token provider and failed to authenticate. - apply(): sasl.aws_msk_iam.enabled now implies SASL on its own; previously it only took effect when the legacy sasl.enabled was also set, so MSK IAM silently stayed on plaintext/ssl. Add a KafkaAuthConfig::apply test matrix covering the full auth surface mapped to the MSK port matrix (plaintext :9092, TLS/mTLS :9094, SASL/SCRAM :9096, SASL/IAM :9098) across the TLS/non-TLS axis, plus a source-context regression test that the OAUTHBEARER refresh callback is enabled and delegated. Signed-off-by: Gaál György --- src/kafka.rs | 230 +++++++++++++++++++++++++++++++++++++++++-- src/sources/kafka.rs | 43 ++++++++ 2 files changed, 263 insertions(+), 10 deletions(-) diff --git a/src/kafka.rs b/src/kafka.rs index 668a10385c2d8..74b9dea37381b 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -128,7 +128,19 @@ pub struct AwsMskIamConfig { impl KafkaAuthConfig { pub(crate) fn apply(&self, client: &mut ClientConfig) -> crate::Result<()> { - let sasl_enabled = self.sasl.as_ref().and_then(|s| s.enabled).unwrap_or(false); + #[cfg(feature = "aws-core")] + let msk_iam_enabled = self + .sasl + .as_ref() + .and_then(|s| s.aws_msk_iam.as_ref()) + .map(|c| c.enabled) + .unwrap_or(false); + #[cfg(not(feature = "aws-core"))] + let msk_iam_enabled = false; + + // MSK IAM (SASL/OAUTHBEARER) implies SASL even if the legacy `sasl.enabled` flag is unset. + let sasl_enabled = + self.sasl.as_ref().and_then(|s| s.enabled).unwrap_or(false) || msk_iam_enabled; let tls_enabled = self.tls.as_ref().and_then(|s| s.enabled).unwrap_or(false); let protocol = match (sasl_enabled, tls_enabled) { @@ -142,15 +154,6 @@ impl KafkaAuthConfig { if sasl_enabled { let sasl = self.sasl.as_ref().unwrap(); - #[cfg(feature = "aws-core")] - let msk_iam_enabled = sasl - .aws_msk_iam - .as_ref() - .map(|c| c.enabled) - .unwrap_or(false); - #[cfg(not(feature = "aws-core"))] - let msk_iam_enabled = false; - if msk_iam_enabled { // OAUTHBEARER: the token itself is produced by // `KafkaStatisticsContext::generate_oauth_token`. No username/password. @@ -356,3 +359,210 @@ impl rdkafka::producer::ProducerContext for KafkaStatisticsContext { ) { } } + +#[cfg(test)] +mod auth_tests { + //! Exercises `KafkaAuthConfig::apply` across the full Kafka authentication matrix, mapped to the + //! AWS MSK connection options (docs: MSK "Port information"): + //! * PLAINTEXT (:9092) -> `security.protocol=plaintext` + //! * TLS encryption / mTLS (:9094) -> `security.protocol=ssl` + //! * SASL/SCRAM over TLS (:9096) -> `security.protocol=sasl_ssl`, mechanism SCRAM + //! * SASL/IAM over TLS (:9098) -> `security.protocol=sasl_ssl`, mechanism OAUTHBEARER + //! plus the non-MSK-but-valid librdkafka combos (SASL over plaintext) and the TLS<->non-TLS axis. + use super::*; + use rdkafka::ClientConfig; + + fn applied(auth: &KafkaAuthConfig) -> ClientConfig { + let mut cc = ClientConfig::new(); + auth.apply(&mut cc).expect("apply"); + cc + } + + fn sasl(enabled: Option, mechanism: Option<&str>, user: Option<&str>) -> KafkaSaslConfig { + KafkaSaslConfig { + enabled, + username: user.map(str::to_owned), + password: user.map(|_| "pw".to_owned().into()), + mechanism: mechanism.map(str::to_owned), + ..Default::default() + } + } + + fn tls(enabled: bool) -> TlsEnableableConfig { + TlsEnableableConfig { + enabled: Some(enabled), + ..Default::default() + } + } + + // ---- security.protocol matrix: (sasl_enabled x tls_enabled) ---- + + #[test] + fn protocol_plaintext_when_nothing_enabled() { + let cc = applied(&KafkaAuthConfig::default()); + assert_eq!(cc.get("security.protocol"), Some("plaintext")); + assert_eq!(cc.get("sasl.mechanism"), None); + } + + #[test] + fn protocol_ssl_when_only_tls() { + // MSK TLS encryption / mTLS listener (:9094). + let auth = KafkaAuthConfig { + sasl: None, + tls: Some(tls(true)), + }; + let cc = applied(&auth); + assert_eq!(cc.get("security.protocol"), Some("ssl")); + assert_eq!(cc.get("sasl.mechanism"), None); + } + + #[test] + fn protocol_sasl_plaintext_when_sasl_without_tls() { + // Not an MSK option, but a valid librdkafka combo for self-managed Kafka. + let auth = KafkaAuthConfig { + sasl: Some(sasl(Some(true), Some("SCRAM-SHA-512"), Some("u"))), + tls: None, + }; + assert_eq!(applied(&auth).get("security.protocol"), Some("sasl_plaintext")); + } + + #[test] + fn protocol_sasl_ssl_when_sasl_and_tls() { + let auth = KafkaAuthConfig { + sasl: Some(sasl(Some(true), Some("SCRAM-SHA-512"), Some("u"))), + tls: Some(tls(true)), + }; + assert_eq!(applied(&auth).get("security.protocol"), Some("sasl_ssl")); + } + + // ---- SASL mechanism selection (non-MSK) ---- + + #[test] + fn scram_sets_mechanism_and_credentials() { + // MSK SASL/SCRAM listener (:9096). + let auth = KafkaAuthConfig { + sasl: Some(sasl(Some(true), Some("SCRAM-SHA-512"), Some("alice"))), + tls: Some(tls(true)), + }; + let cc = applied(&auth); + assert_eq!(cc.get("sasl.mechanism"), Some("SCRAM-SHA-512")); + assert_eq!(cc.get("sasl.username"), Some("alice")); + assert_eq!(cc.get("sasl.password"), Some("pw")); + } + + #[test] + fn plain_mechanism_is_passed_through() { + let auth = KafkaAuthConfig { + sasl: Some(sasl(Some(true), Some("PLAIN"), Some("alice"))), + tls: Some(tls(true)), + }; + assert_eq!(applied(&auth).get("sasl.mechanism"), Some("PLAIN")); + } + + #[test] + fn sasl_present_but_disabled_sets_no_mechanism() { + let auth = KafkaAuthConfig { + sasl: Some(sasl(Some(false), Some("SCRAM-SHA-512"), Some("u"))), + tls: Some(tls(true)), + }; + let cc = applied(&auth); + assert_eq!(cc.get("security.protocol"), Some("ssl")); // sasl off -> just TLS + assert_eq!(cc.get("sasl.mechanism"), None); + } + + // ---- TLS option plumbing ---- + + #[test] + fn tls_verify_certificate_flag_is_plumbed() { + let mut t = tls(true); + t.options.verify_certificate = Some(false); + let auth = KafkaAuthConfig { + sasl: None, + tls: Some(t), + }; + assert_eq!( + applied(&auth).get("enable.ssl.certificate.verification"), + Some("false") + ); + } + + // ---- MSK IAM (SASL/OAUTHBEARER) ---- + + #[cfg(feature = "aws-core")] + fn msk_sasl(sasl_enabled: Option, user: Option<&str>) -> KafkaSaslConfig { + KafkaSaslConfig { + enabled: sasl_enabled, + username: user.map(str::to_owned), + password: user.map(|_| "pw".to_owned().into()), + mechanism: None, + aws_msk_iam: Some(AwsMskIamConfig { + enabled: true, + ..Default::default() + }), + } + } + + #[cfg(feature = "aws-core")] + #[test] + fn msk_iam_over_tls_uses_oauthbearer() { + // MSK SASL/IAM listener (:9098): SASL_SSL + OAUTHBEARER, no username/password. + let auth = KafkaAuthConfig { + sasl: Some(msk_sasl(Some(true), None)), + tls: Some(tls(true)), + }; + let cc = applied(&auth); + assert_eq!(cc.get("security.protocol"), Some("sasl_ssl")); + assert_eq!(cc.get("sasl.mechanism"), Some("OAUTHBEARER")); + assert_eq!(cc.get("sasl.username"), None); + assert_eq!(cc.get("sasl.password"), None); + } + + #[cfg(feature = "aws-core")] + #[test] + fn msk_iam_enables_sasl_without_legacy_flag() { + // Regression: `aws_msk_iam.enabled` alone must turn SASL on (sasl.enabled left unset). + let auth = KafkaAuthConfig { + sasl: Some(msk_sasl(None, None)), + tls: Some(tls(true)), + }; + let cc = applied(&auth); + assert_eq!(cc.get("security.protocol"), Some("sasl_ssl")); + assert_eq!(cc.get("sasl.mechanism"), Some("OAUTHBEARER")); + } + + #[cfg(feature = "aws-core")] + #[test] + fn msk_iam_ignores_username_password() { + // If both MSK IAM and username/password are set, OAUTHBEARER wins and no creds are emitted. + let auth = KafkaAuthConfig { + sasl: Some(msk_sasl(Some(true), Some("alice"))), + tls: Some(tls(true)), + }; + let cc = applied(&auth); + assert_eq!(cc.get("sasl.mechanism"), Some("OAUTHBEARER")); + assert_eq!(cc.get("sasl.username"), None); + assert_eq!(cc.get("sasl.password"), None); + } + + #[cfg(feature = "aws-core")] + #[test] + fn msk_iam_disabled_flag_falls_back_to_plain_sasl() { + // aws_msk_iam present but enabled=false -> behaves like ordinary SASL config. + let auth = KafkaAuthConfig { + sasl: Some(KafkaSaslConfig { + enabled: Some(true), + mechanism: Some("SCRAM-SHA-256".to_owned()), + username: Some("u".to_owned()), + password: Some("pw".to_owned().into()), + aws_msk_iam: Some(AwsMskIamConfig { + enabled: false, + ..Default::default() + }), + }), + tls: Some(tls(true)), + }; + let cc = applied(&auth); + assert_eq!(cc.get("sasl.mechanism"), Some("SCRAM-SHA-256")); + assert_eq!(cc.get("sasl.username"), Some("u")); + } +} diff --git a/src/sources/kafka.rs b/src/sources/kafka.rs index 924927954d87c..66dd89bd34534 100644 --- a/src/sources/kafka.rs +++ b/src/sources/kafka.rs @@ -1376,9 +1376,24 @@ impl KafkaSourceContext { } impl ClientContext for KafkaSourceContext { + // librdkafka calls into this (the consumer's) context, so the OAUTHBEARER refresh must be + // enabled and forwarded here — not just on the inner `KafkaStatisticsContext` (which only the + // sink uses directly). Without this, MSK IAM sources are configured with + // `sasl.mechanism=OAUTHBEARER` but never get a token provider. + #[cfg(feature = "aws-core")] + const ENABLE_REFRESH_OAUTH_TOKEN: bool = true; + fn stats(&self, statistics: Statistics) { self.stats.stats(statistics) } + + #[cfg(feature = "aws-core")] + fn generate_oauth_token( + &self, + oauthbearer_config: Option<&str>, + ) -> Result> { + self.stats.generate_oauth_token(oauthbearer_config) + } } impl ConsumerContext for KafkaSourceContext { @@ -1421,6 +1436,34 @@ mod test { crate::test_util::test_generate_config::(); } + // Regression: the consumer's rdkafka context is `KafkaSourceContext`, not the inner + // `KafkaStatisticsContext`. It must enable and forward the OAUTHBEARER refresh callback, or MSK + // IAM sources get `sasl.mechanism=OAUTHBEARER` with no token provider and fail to authenticate. + #[cfg(feature = "aws-core")] + #[test] + fn source_context_enables_and_forwards_oauth_refresh() { + use rdkafka::ClientContext; + + assert!( + ::ENABLE_REFRESH_OAUTH_TOKEN, + "source context must enable the OAUTHBEARER refresh callback" + ); + + // With no MSK creds, the call must surface OUR error (proving it delegates to the inner + // KafkaStatisticsContext) rather than rdkafka's default "must be overridden" error. + let (tx, _rx) = mpsc::unbounded_channel(); + let ctx = KafkaSourceContext::new(false, false, tx, Span::none(), None); + // `OAuthToken` is not `Debug`, so match instead of `expect_err`. + let err = match ctx.generate_oauth_token(None) { + Ok(_) => panic!("should error without MSK creds"), + Err(e) => e.to_string(), + }; + assert!( + err.contains("MSK IAM is not configured"), + "expected delegated MSK error, got: {err}" + ); + } + pub(super) fn make_config( topic: &str, group: &str, From e60c624409cc863e48547c2996b0452453e273f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Mon, 6 Jul 2026 16:27:06 +0200 Subject: [PATCH 03/10] fix(kafka): make MSK IAM token callback sync, require TLS, poll healthcheck MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second round of review fixes: - Nested-runtime panic: the source serves the OAUTHBEARER refresh callback while the StreamConsumer is polled inside the tokio runtime, so the previous Handle::block_on in generate_oauth_token would panic. Token minting is now fully synchronous: AWS credentials are resolved once and cached in an Arc> that a background task refreshes before expiry (Weak ref, so it stops when the component drops); the callback only reads the cache and signs. Removes the runtime handle from the context entirely. - Healthcheck: switch the sink healthcheck from BaseProducer (never polled) to FutureProducer, whose background poll thread serves the token-refresh event so the initial MSK IAM token is installed before fetch_metadata. - Require TLS: reject sasl.aws_msk_iam without tls.enabled instead of silently negotiating sasl_plaintext, which MSK IAM's SASL_SSL listener rejects. - Fix the KafkaSaslConfig literal in the sink integration tests so the aws-core + kafka-integration-tests build compiles. Add tests: refresh skew/floor, credential-cache path, and MSK-IAM-without-TLS rejection. Signed-off-by: Gaál György --- src/kafka.rs | 75 +++++++++-------- src/kafka/msk_iam.rs | 176 +++++++++++++++++++++++---------------- src/sinks/kafka/sink.rs | 7 +- src/sinks/kafka/tests.rs | 1 + 4 files changed, 150 insertions(+), 109 deletions(-) diff --git a/src/kafka.rs b/src/kafka.rs index 74b9dea37381b..71800573ce787 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -15,14 +15,12 @@ use crate::{ mod msk_iam; #[cfg(feature = "aws-core")] -use aws_credential_types::provider::SharedCredentialsProvider; +use std::time::SystemTime; #[cfg(feature = "aws-core")] use aws_types::region::Region; #[cfg(feature = "aws-core")] use rdkafka::client::OAuthToken; #[cfg(feature = "aws-core")] -use tokio::runtime::Handle; -#[cfg(feature = "aws-core")] use crate::{ aws::{AwsAuthentication, RegionOrEndpoint}, config::ProxyConfig, @@ -143,6 +141,16 @@ impl KafkaAuthConfig { self.sasl.as_ref().and_then(|s| s.enabled).unwrap_or(false) || msk_iam_enabled; let tls_enabled = self.tls.as_ref().and_then(|s| s.enabled).unwrap_or(false); + // MSK IAM is only offered on the SASL_SSL (IAM) listener, so TLS is mandatory. Reject early + // rather than silently negotiating `sasl_plaintext` and failing to connect. + if msk_iam_enabled && !tls_enabled { + return Err( + "`sasl.aws_msk_iam` requires TLS; set `tls.enabled = true` (MSK IAM uses the \ + SASL_SSL port, 9098)" + .into(), + ); + } + let protocol = match (sasl_enabled, tls_enabled) { (false, false) => "plaintext", (false, true) => "ssl", @@ -244,17 +252,20 @@ impl KafkaAuthConfig { .auth .credentials_provider(region.clone(), proxy, None) .await?; - Ok(Some(MskIamCredentials { region, provider })) + // Resolve now (fail fast on misconfig) and keep them refreshed in the background so the + // token callback stays synchronous. + let credentials = msk_iam::spawn_credentials_cache(provider).await?; + Ok(Some(MskIamCredentials { region, credentials })) } } -/// AWS credential material captured at build time and moved into the rdkafka client context so the -/// (synchronous) OAUTHBEARER refresh callback can mint fresh tokens. +/// AWS credential material moved into the rdkafka client context so the (synchronous) OAUTHBEARER +/// refresh callback can mint fresh tokens without any async/`block_on`. #[cfg(feature = "aws-core")] #[derive(Clone)] pub(crate) struct MskIamCredentials { pub(crate) region: Region, - pub(crate) provider: SharedCredentialsProvider, + pub(crate) credentials: msk_iam::CredentialsCache, } fn pathbuf_to_string(path: &Path) -> crate::Result<&str> { @@ -268,11 +279,6 @@ pub(crate) struct KafkaStatisticsContext { /// Present only when MSK IAM (SASL/OAUTHBEARER) auth is enabled. #[cfg(feature = "aws-core")] pub(crate) msk_iam: Option, - /// Handle to the tokio runtime, captured at construction. The OAUTHBEARER callback runs on - /// librdkafka's own (non-tokio) thread, so it bridges to async credential resolution via - /// `Handle::block_on`. `Option` because construction may happen outside a runtime (e.g. tests). - #[cfg(feature = "aws-core")] - pub(crate) runtime: Option, } impl KafkaStatisticsContext { @@ -286,7 +292,6 @@ impl KafkaStatisticsContext { expose_lag_metrics, span, msk_iam, - runtime: Handle::try_current().ok(), } } @@ -324,16 +329,15 @@ impl ClientContext for KafkaStatisticsContext { .msk_iam .as_ref() .ok_or("generate_oauth_token called but MSK IAM is not configured")?; - let runtime = self - .runtime - .as_ref() - .ok_or("no tokio runtime handle available to mint the MSK IAM token")?; - // The MSK signer signs a synthetic `kafka.{region}.amazonaws.com` host, so region + - // credentials are all the callback needs. block_on is safe: this runs on librdkafka's own - // (non-tokio) thread, so there is no runtime nesting. - let token = - runtime.block_on(msk_iam::generate_auth_token(&creds.region, &creds.provider))?; + // Fully synchronous: read the background-refreshed credentials and sign. No async / block_on, + // so this is safe even when invoked on the tokio runtime thread that polls the consumer. + let credentials = creds + .credentials + .lock() + .map_err(|_| "MSK IAM credential cache mutex poisoned")? + .clone(); + let token = msk_iam::build_token(&creds.region, &credentials, SystemTime::now())?; Ok(OAuthToken { token: token.token, @@ -346,20 +350,6 @@ impl ClientContext for KafkaStatisticsContext { impl ConsumerContext for KafkaStatisticsContext {} -// Enables `KafkaStatisticsContext` to back a `BaseProducer` directly (used by the sink healthcheck), -// so the OAUTHBEARER token callback is available there too. The sink's `FutureProducer` wraps this -// context separately and is unaffected. -impl rdkafka::producer::ProducerContext for KafkaStatisticsContext { - type DeliveryOpaque = (); - - fn delivery( - &self, - _delivery_result: &rdkafka::producer::DeliveryResult<'_>, - _delivery_opaque: Self::DeliveryOpaque, - ) { - } -} - #[cfg(test)] mod auth_tests { //! Exercises `KafkaAuthConfig::apply` across the full Kafka authentication matrix, mapped to the @@ -517,6 +507,19 @@ mod auth_tests { assert_eq!(cc.get("sasl.password"), None); } + #[cfg(feature = "aws-core")] + #[test] + fn msk_iam_without_tls_is_rejected() { + // MSK IAM is SASL_SSL-only; applying without TLS must be a hard error, not sasl_plaintext. + let auth = KafkaAuthConfig { + sasl: Some(msk_sasl(Some(true), None)), + tls: None, + }; + let mut cc = ClientConfig::new(); + let err = auth.apply(&mut cc).expect_err("must reject MSK IAM without TLS"); + assert!(err.to_string().contains("requires TLS"), "got: {err}"); + } + #[cfg(feature = "aws-core")] #[test] fn msk_iam_enables_sasl_without_legacy_flag() { diff --git a/src/kafka/msk_iam.rs b/src/kafka/msk_iam.rs index 4edae7c86952d..572236afcd684 100644 --- a/src/kafka/msk_iam.rs +++ b/src/kafka/msk_iam.rs @@ -5,7 +5,14 @@ //! `kafka-cluster:Connect` action, base64url-encoded. This is a port of the official //! `aws-msk-iam-sasl-signer-go` `constructAuthToken`; correctness is proven offline in the tests //! below against an independent SigV4 oracle (fixed inputs -> known-answer signature). +//! +//! Token minting is fully synchronous: `build_token` only signs (no I/O, no async). AWS credential +//! resolution — the one async step — is done ahead of time and cached in an `Arc>` +//! that a background task refreshes before expiry. This keeps `generate_oauth_token` free of any +//! `block_on`, which is essential because for the source it is invoked while the `StreamConsumer` is +//! polled *inside* the tokio runtime, where nested blocking would panic. +use std::sync::{Arc, Mutex, Weak}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use aws_credential_types::Credentials; @@ -20,11 +27,20 @@ use base64::Engine as _; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use snafu::{ResultExt, Snafu}; +/// MSK IAM presigned tokens are valid for at most 15 minutes. librdkafka refreshes at ~80% of the +/// remaining lifetime, so 900s gives a comfortable refresh cadence. const EXPIRY_SECONDS: u64 = 900; const SIGNING_NAME: &str = "kafka-cluster"; const ACTION: &str = "kafka-cluster:Connect"; const USER_AGENT: &str = concat!("vector-msk-iam-signer/", env!("CARGO_PKG_VERSION")); +/// Refresh cached credentials this long before they expire. +const CRED_REFRESH_SKEW: Duration = Duration::from_secs(5 * 60); +/// Floor for the refresh sleep, so a near-expired/expired credential doesn't spin. +const CRED_REFRESH_MIN: Duration = Duration::from_secs(60); +/// Cadence used when credentials advertise no expiry (e.g. static keys). +const CRED_REFRESH_NO_EXPIRY: Duration = Duration::from_secs(15 * 60); + type BoxError = Box; #[derive(Debug, Snafu)] @@ -39,6 +55,9 @@ pub enum MskIamError { Signing { source: BoxError }, } +/// Shared, background-refreshed AWS credentials used to mint MSK IAM tokens. +pub(crate) type CredentialsCache = Arc>; + /// Token librdkafka hands to the broker, plus its absolute expiry (unix-epoch ms). #[derive(Debug, Clone)] pub struct MskAuthToken { @@ -46,22 +65,52 @@ pub struct MskAuthToken { pub lifetime_ms: i64, } -/// Generate a fresh MSK IAM SASL/OAUTHBEARER token for `region`. -pub async fn generate_auth_token( - region: &Region, - credentials_provider: &SharedCredentialsProvider, -) -> Result { - let credentials = credentials_provider - .provide_credentials() - .await - .context(CredentialsSnafu)?; - let signing_time = SystemTime::now(); - build_token(region, &credentials, signing_time) +/// Resolve credentials once (async), then spawn a background task that keeps them fresh, and return +/// the shared cache. The `generate_oauth_token` callback reads this cache synchronously. +/// +/// The refresher holds only a `Weak` reference, so it exits automatically once the owning +/// component (and thus the last `CredentialsCache` clone) is dropped. +pub(crate) async fn spawn_credentials_cache( + provider: SharedCredentialsProvider, +) -> Result { + let initial = provider.provide_credentials().await.context(CredentialsSnafu)?; + let cache: CredentialsCache = Arc::new(Mutex::new(initial)); + tokio::spawn(refresh_loop(provider, Arc::downgrade(&cache))); + Ok(cache) +} + +async fn refresh_loop(provider: SharedCredentialsProvider, weak: Weak>) { + loop { + let sleep_for = match weak.upgrade() { + None => break, + Some(cache) => next_refresh(cache.lock().ok().and_then(|c| c.expiry())), + }; + tokio::time::sleep(sleep_for).await; + + let Some(cache) = weak.upgrade() else { break }; + // On failure keep the previous credentials and retry on the next iteration. + if let Ok(fresh) = provider.provide_credentials().await { + if let Ok(mut guard) = cache.lock() { + *guard = fresh; + } + } + } +} + +fn next_refresh(expiry: Option) -> Duration { + match expiry { + Some(exp) => exp + .duration_since(SystemTime::now()) + .unwrap_or(Duration::ZERO) + .saturating_sub(CRED_REFRESH_SKEW) + .max(CRED_REFRESH_MIN), + None => CRED_REFRESH_NO_EXPIRY, + } } -/// Deterministic core: given credentials + a fixed signing time, produce the token. Split out so -/// tests can pin the time and assert a known-answer signature. -fn build_token( +/// Mint a fresh MSK IAM SASL/OAUTHBEARER token for `region` from already-resolved `credentials`. +/// Fully synchronous — safe to call from librdkafka's refresh callback on any thread. +pub(crate) fn build_token( region: &Region, credentials: &Credentials, signing_time: SystemTime, @@ -163,6 +212,14 @@ mod tests { UNIX_EPOCH + Duration::from_secs(FIXED_EPOCH_SECS) } + fn creds(session: bool) -> Credentials { + Credentials::from_keys( + ACCESS_KEY, + SECRET_KEY, + session.then(|| SESSION_TOKEN.to_string()), + ) + } + fn query_param<'a>(url: &'a str, key: &str) -> Option<&'a str> { let q = url.split_once('?')?.1; q.split('&').find_map(|kv| { @@ -186,56 +243,37 @@ mod tests { (base.to_string(), params) } - // Full presigned URLs from the independent oracle (oracle/sigv4_oracle.py output). + // Full presigned URL from the independent oracle (oracle/sigv4_oracle.py output). const ORACLE_URL_WITH_TOKEN: &str = "https://kafka.ap-southeast-1.amazonaws.com/?Action=kafka-cluster%3AConnect&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=AKIDEXAMPLE%2F20231114%2Fap-southeast-1%2Fkafka-cluster%2Faws4_request&X-Amz-Date=20231114T221320Z&X-Amz-Expires=900&X-Amz-Security-Token=IQoJb3Jp&X-Amz-SignedHeaders=host&X-Amz-Signature=4c02acd7700d674cecd2b27f6a906be056df725db5c7571700fd54a354a31904"; - /// STRONGEST PROOF: the *entire* presigned URL (every param, every encoded value, the signature) - /// matches the independent oracle — not just the signature. Order-independent because MSK - /// re-canonicalizes the query on its side. + /// STRONGEST PROOF: the entire presigned URL (every param, every encoded value, the signature) + /// matches the independent oracle — order-independent because MSK re-canonicalizes the query. #[test] fn full_presigned_url_matches_oracle() { - let creds = - Credentials::from_keys(ACCESS_KEY, SECRET_KEY, Some(SESSION_TOKEN.to_string())); - let url = presign(®ion(), &creds, fixed_time()).expect("presign"); + let url = presign(®ion(), &creds(true), fixed_time()).expect("presign"); assert_eq!(normalize(&url), normalize(ORACLE_URL_WITH_TOKEN), "url={url}"); } - /// KNOWN-ANSWER: the aws-sigv4 signature must equal the independent oracle's, byte-for-byte, - /// with a session token present (X-Amz-Security-Token in the signed canonical query). + /// KNOWN-ANSWER: aws-sigv4 signature must equal the oracle's, with a session token present. #[test] fn signature_matches_oracle_with_session_token() { - let creds = - Credentials::from_keys(ACCESS_KEY, SECRET_KEY, Some(SESSION_TOKEN.to_string())); - let url = presign(®ion(), &creds, fixed_time()).expect("presign"); - - assert_eq!( - query_param(&url, "X-Amz-Signature"), - Some(GOLDEN_SIG_WITH_TOKEN), - "aws-sigv4 signature diverged from the independent SigV4 oracle\nurl={url}" - ); + let url = presign(®ion(), &creds(true), fixed_time()).expect("presign"); + assert_eq!(query_param(&url, "X-Amz-Signature"), Some(GOLDEN_SIG_WITH_TOKEN)); assert_eq!(query_param(&url, "X-Amz-Security-Token"), Some("IQoJb3Jp")); } /// KNOWN-ANSWER: same, without a session token (static long-term credentials). #[test] fn signature_matches_oracle_without_session_token() { - let creds = Credentials::from_keys(ACCESS_KEY, SECRET_KEY, None); - let url = presign(®ion(), &creds, fixed_time()).expect("presign"); - - assert_eq!( - query_param(&url, "X-Amz-Signature"), - Some(GOLDEN_SIG_NO_TOKEN), - "aws-sigv4 signature diverged from the independent SigV4 oracle\nurl={url}" - ); + let url = presign(®ion(), &creds(false), fixed_time()).expect("presign"); + assert_eq!(query_param(&url, "X-Amz-Signature"), Some(GOLDEN_SIG_NO_TOKEN)); assert_eq!(query_param(&url, "X-Amz-Security-Token"), None); } /// Structural invariants the MSK broker requires of the presigned URL. #[test] fn presigned_url_has_required_msk_shape() { - let creds = Credentials::from_keys(ACCESS_KEY, SECRET_KEY, None); - let url = presign(®ion(), &creds, fixed_time()).expect("presign"); - + let url = presign(®ion(), &creds(false), fixed_time()).expect("presign"); assert!(url.starts_with("https://kafka.ap-southeast-1.amazonaws.com/?")); assert_eq!(query_param(&url, "Action"), Some("kafka-cluster%3AConnect")); assert_eq!(query_param(&url, "X-Amz-Algorithm"), Some("AWS4-HMAC-SHA256")); @@ -247,45 +285,41 @@ mod tests { ); } - /// The final token is base64url-nopad of the presigned URL (+ appended User-Agent). + /// The token is base64url-nopad of the presigned URL (+ appended User-Agent), and its lifetime + /// is the absolute unix-epoch ms of signing-time + 900s. #[test] - fn token_is_base64url_nopad_of_presigned_url() { - let creds = Credentials::from_keys(ACCESS_KEY, SECRET_KEY, None); - let out = build_token(®ion(), &creds, fixed_time()).expect("token"); - + fn build_token_encodes_and_sets_lifetime() { + let out = build_token(®ion(), &creds(false), fixed_time()).expect("token"); assert!(!out.token.contains('='), "token must be unpadded base64url"); let decoded = URL_SAFE_NO_PAD.decode(out.token.as_bytes()).expect("decode"); let url = String::from_utf8(decoded).expect("utf8"); assert!(url.starts_with("https://kafka.ap-southeast-1.amazonaws.com/?")); assert!(url.ends_with(&format!("&User-Agent={USER_AGENT}"))); assert!(url.contains(GOLDEN_SIG_NO_TOKEN)); - - // Absolute expiry = signing time + 900s, in unix-epoch ms. assert_eq!(out.lifetime_ms, (FIXED_EPOCH_SECS as i64) * 1000 + 900_000); } - /// End-to-end through the public async entrypoint with a static credentials provider. + /// The cache path: resolve via a static provider, then mint a token from the cached credentials. #[tokio::test] - async fn generate_auth_token_via_provider() { - let provider = SharedCredentialsProvider::new(Credentials::from_keys( - ACCESS_KEY, - SECRET_KEY, - Some(SESSION_TOKEN.to_string()), - )); - let out = generate_auth_token(®ion(), &provider) - .await - .expect("generate"); + async fn credentials_cache_then_build_token() { + let provider = SharedCredentialsProvider::new(creds(true)); + let cache = spawn_credentials_cache(provider).await.expect("cache"); + let cached = cache.lock().unwrap().clone(); + let out = build_token(®ion(), &cached, fixed_time()).expect("token"); + let decoded = URL_SAFE_NO_PAD.decode(out.token.as_bytes()).unwrap(); + let url = String::from_utf8(decoded).unwrap(); + assert!(url.contains(GOLDEN_SIG_WITH_TOKEN)); + } - let decoded = URL_SAFE_NO_PAD.decode(out.token.as_bytes()).expect("decode"); - let url = String::from_utf8(decoded).expect("utf8"); - assert!(url.contains("Action=kafka-cluster%3AConnect")); - assert!(url.contains("X-Amz-Signature=")); - // Uses wall-clock now(): lifetime must be ~900s in the future, so comfortably positive. - let now_ms = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_millis() as i64; - assert!(out.lifetime_ms > now_ms); - assert!(out.lifetime_ms <= now_ms + 900_000 + 5_000); + #[test] + fn refresh_uses_skew_and_floor() { + // Expiry far out -> refresh ~skew before it. + let far = SystemTime::now() + Duration::from_secs(3600); + assert!(next_refresh(Some(far)) >= Duration::from_secs(3600 - 5 * 60 - 5)); + // Already expired -> clamped to the floor, never zero. + let past = SystemTime::now() - Duration::from_secs(10); + assert_eq!(next_refresh(Some(past)), CRED_REFRESH_MIN); + // No expiry -> loose cadence. + assert_eq!(next_refresh(None), CRED_REFRESH_NO_EXPIRY); } } diff --git a/src/sinks/kafka/sink.rs b/src/sinks/kafka/sink.rs index e6b1eb91892ac..26ee03622f65d 100644 --- a/src/sinks/kafka/sink.rs +++ b/src/sinks/kafka/sink.rs @@ -3,7 +3,7 @@ use std::time::Duration; use rdkafka::{ ClientConfig, error::KafkaError, - producer::{BaseProducer, FutureProducer, Producer}, + producer::{FutureProducer, Producer}, }; use snafu::{ResultExt, Snafu}; use tower::limit::RateLimit; @@ -150,7 +150,10 @@ pub(crate) async fn healthcheck( let context = KafkaStatisticsContext::new(false, Span::current()); tokio::task::spawn_blocking(move || { - let producer: BaseProducer<_> = client_config.create_with_context(context).unwrap(); + // A `FutureProducer` runs its own background poll thread, which serves the OAUTHBEARER token + // refresh event; a `BaseProducer` would need an explicit `poll()` and the MSK IAM token + // would never be installed before `fetch_metadata`. + let producer: FutureProducer<_> = client_config.create_with_context(context).unwrap(); let topic = topic.as_deref(); producer diff --git a/src/sinks/kafka/tests.rs b/src/sinks/kafka/tests.rs index ab72f3402b630..8b7642bddbaad 100644 --- a/src/sinks/kafka/tests.rs +++ b/src/sinks/kafka/tests.rs @@ -323,6 +323,7 @@ mod integration_test { username: Some("admin".to_string()), password: Some("admin".to_string().into()), mechanism: Some("PLAIN".to_owned()), + ..Default::default() }), None, KafkaCompression::None, From 014b6ca5eb6d6b25667b2d01c53d30553e5fea35 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Mon, 6 Jul 2026 18:32:55 +0200 Subject: [PATCH 04/10] fix(kafka): reject bare OAUTHBEARER mechanism outside MSK IAM MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rust-rdkafka gates the OAUTHBEARER refresh callback on a compile-time associated const (ClientContext::ENABLE_REFRESH_OAUTH_TOKEN), so in aws-core builds it is enabled for the context whenever it exists; it cannot be toggled per-instance by runtime config without monomorphizing the whole consumer/producer path over both values. Vector only provides a token for OAUTHBEARER via MSK IAM, so a bare sasl.mechanism=OAUTHBEARER (without sasl.aws_msk_iam) is now rejected at config time with guidance, instead of failing later in the refresh callback with no token provider. Add a regression test. Signed-off-by: Gaál György --- src/kafka.rs | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/src/kafka.rs b/src/kafka.rs index 71800573ce787..7e9104d80304b 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -151,6 +151,25 @@ impl KafkaAuthConfig { ); } + // rust-rdkafka gates the OAUTHBEARER refresh callback on a compile-time associated const, so + // in `aws-core` builds it is enabled for the context whenever it exists. Vector only provides + // a token for it via MSK IAM. Reject a bare `OAUTHBEARER` mechanism up front with guidance, + // rather than letting the callback fire later with no token provider. + #[cfg(feature = "aws-core")] + if !msk_iam_enabled + && self + .sasl + .as_ref() + .and_then(|s| s.mechanism.as_deref()) + .is_some_and(|m| m.eq_ignore_ascii_case("OAUTHBEARER")) + { + return Err( + "`sasl.mechanism = \"OAUTHBEARER\"` is only supported via `sasl.aws_msk_iam` \ + (AWS MSK IAM)" + .into(), + ); + } + let protocol = match (sasl_enabled, tls_enabled) { (false, false) => "plaintext", (false, true) => "ssl", @@ -507,6 +526,19 @@ mod auth_tests { assert_eq!(cc.get("sasl.password"), None); } + #[cfg(feature = "aws-core")] + #[test] + fn bare_oauthbearer_mechanism_is_rejected() { + // OAUTHBEARER is only wired up for MSK IAM; a manual mechanism must fail fast with guidance. + let auth = KafkaAuthConfig { + sasl: Some(sasl(Some(true), Some("OAUTHBEARER"), None)), + tls: Some(tls(true)), + }; + let mut cc = ClientConfig::new(); + let err = auth.apply(&mut cc).expect_err("bare OAUTHBEARER must be rejected"); + assert!(err.to_string().contains("aws_msk_iam"), "got: {err}"); + } + #[cfg(feature = "aws-core")] #[test] fn msk_iam_without_tls_is_rejected() { From 6b38578dd7d00265bc1d2de43aca32f59361f078 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Mon, 6 Jul 2026 19:23:52 +0200 Subject: [PATCH 05/10] fix(kafka): honor MSK IAM load timeout, gate OAUTHBEARER guard on sasl, reject conflicting librdkafka_options MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third round of review fixes: - Credential load timeout: MSK IAM credential resolution now honors the configured load_timeout_secs (5s default) via a new AwsAuthentication::load_timeout() accessor, wrapping both the initial resolve and the background refresh in a timeout so a stalled IMDS/STS/profile lookup can't hang the component build. - The bare-OAUTHBEARER rejection is now gated on sasl_enabled, so an inactive SASL stanza (enabled=false) kept around for templating/rollbacks is ignored as before. - Reject librdkafka_options that would override MSK IAM's security.protocol / sasl.* (both source and sink apply librdkafka_options after auth.apply, which would otherwise silently drop OAUTHBEARER/SASL_SSL). Add tests for the load timeout, disabled-stanza pass-through, and conflicting/benign librdkafka_options. Signed-off-by: Gaál György --- src/aws/auth.rs | 17 +++++++ src/kafka.rs | 98 +++++++++++++++++++++++++++++++++++++-- src/kafka/msk_iam.rs | 62 ++++++++++++++++++++++--- src/sinks/kafka/config.rs | 6 +++ src/sources/kafka.rs | 6 +++ 5 files changed, 179 insertions(+), 10 deletions(-) diff --git a/src/aws/auth.rs b/src/aws/auth.rs index 7ca3de0285b61..8799d796d0bc9 100644 --- a/src/aws/auth.rs +++ b/src/aws/auth.rs @@ -212,6 +212,23 @@ fn default_profile() -> String { } impl AwsAuthentication { + /// The effective credential load timeout for this authentication mechanism (the configured + /// `load_timeout_secs`, or the shared 5s default). Mirrors the timeout applied by + /// [`AwsAuthentication::credentials_cache`] for components that resolve credentials directly. + pub(crate) fn load_timeout(&self) -> Duration { + match self { + AwsAuthentication::Role { + load_timeout_secs, .. + } + | AwsAuthentication::Default { + load_timeout_secs, .. + } => load_timeout_secs + .map(Duration::from_secs) + .unwrap_or(DEFAULT_LOAD_TIMEOUT), + _ => DEFAULT_LOAD_TIMEOUT, + } + } + /// Creates the identity cache to store credentials based on the authentication mechanism chosen. pub(super) async fn credentials_cache(&self) -> crate::Result { match self { diff --git a/src/kafka.rs b/src/kafka.rs index 7e9104d80304b..bc6105c58bfb2 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -156,7 +156,8 @@ impl KafkaAuthConfig { // a token for it via MSK IAM. Reject a bare `OAUTHBEARER` mechanism up front with guidance, // rather than letting the callback fire later with no token provider. #[cfg(feature = "aws-core")] - if !msk_iam_enabled + if sasl_enabled + && !msk_iam_enabled && self .sasl .as_ref() @@ -271,11 +272,48 @@ impl KafkaAuthConfig { .auth .credentials_provider(region.clone(), proxy, None) .await?; - // Resolve now (fail fast on misconfig) and keep them refreshed in the background so the - // token callback stays synchronous. - let credentials = msk_iam::spawn_credentials_cache(provider).await?; + // Resolve now (fail fast on misconfig, honoring the configured load timeout) and keep them + // refreshed in the background so the token callback stays synchronous. + let credentials = + msk_iam::spawn_credentials_cache(provider, cfg.auth.load_timeout()).await?; Ok(Some(MskIamCredentials { region, credentials })) } + + /// Reject `librdkafka_options` that would override the security/mechanism settings MSK IAM + /// depends on. Both the source and sink apply `librdkafka_options` *after* [`Self::apply`], so a + /// leftover `security.protocol`/`sasl.*` override would silently drop OAUTHBEARER/SASL_SSL and + /// the component would load credentials but fail to authenticate. + #[cfg(feature = "aws-core")] + pub(crate) fn validate_librdkafka_overrides<'a>( + &self, + option_keys: impl IntoIterator, + ) -> crate::Result<()> { + let msk_iam_enabled = self + .sasl + .as_ref() + .and_then(|s| s.aws_msk_iam.as_ref()) + .map(|c| c.enabled) + .unwrap_or(false); + if !msk_iam_enabled { + return Ok(()); + } + const RESERVED: [&str; 4] = [ + "security.protocol", + "sasl.mechanism", + "sasl.username", + "sasl.password", + ]; + for key in option_keys { + if RESERVED.iter().any(|r| key.eq_ignore_ascii_case(r)) { + return Err(format!( + "`librdkafka_options.{key}` conflicts with `sasl.aws_msk_iam`; remove it \ + (MSK IAM manages `security.protocol` and `sasl.*`)" + ) + .into()); + } + } + Ok(()) + } } /// AWS credential material moved into the rdkafka client context so the (synchronous) OAUTHBEARER @@ -539,6 +577,58 @@ mod auth_tests { assert!(err.to_string().contains("aws_msk_iam"), "got: {err}"); } + #[cfg(feature = "aws-core")] + #[test] + fn disabled_sasl_stanza_with_oauthbearer_is_ignored() { + // An inactive SASL stanza (enabled=false) never sets a mechanism, so it must not be rejected + // — supports keeping a stanza around for templating/rollbacks. + let auth = KafkaAuthConfig { + sasl: Some(sasl(Some(false), Some("OAUTHBEARER"), None)), + tls: Some(tls(true)), + }; + let cc = applied(&auth); + assert_eq!(cc.get("security.protocol"), Some("ssl")); + assert_eq!(cc.get("sasl.mechanism"), None); + } + + #[cfg(feature = "aws-core")] + #[test] + fn librdkafka_overrides_conflicting_with_msk_iam_are_rejected() { + use std::collections::HashMap; + let auth = KafkaAuthConfig { + sasl: Some(msk_sasl(Some(true), None)), + tls: Some(tls(true)), + }; + for key in [ + "security.protocol", + "sasl.mechanism", + "sasl.username", + "sasl.password", + ] { + let opts = HashMap::from([(key.to_string(), "x".to_string())]); + let err = auth + .validate_librdkafka_overrides(opts.keys()) + .expect_err("conflicting override must be rejected"); + assert!(err.to_string().contains(key), "got: {err}"); + } + // Benign options are fine. + let ok = HashMap::from([("socket.timeout.ms".to_string(), "1000".to_string())]); + assert!(auth.validate_librdkafka_overrides(ok.keys()).is_ok()); + } + + #[cfg(feature = "aws-core")] + #[test] + fn librdkafka_overrides_ignored_without_msk_iam() { + // Without MSK IAM these keys are the user's own business. + use std::collections::HashMap; + let auth = KafkaAuthConfig { + sasl: Some(sasl(Some(true), Some("SCRAM-SHA-512"), Some("u"))), + tls: Some(tls(true)), + }; + let opts = HashMap::from([("sasl.mechanism".to_string(), "PLAIN".to_string())]); + assert!(auth.validate_librdkafka_overrides(opts.keys()).is_ok()); + } + #[cfg(feature = "aws-core")] #[test] fn msk_iam_without_tls_is_rejected() { diff --git a/src/kafka/msk_iam.rs b/src/kafka/msk_iam.rs index 572236afcd684..5acef1d5a61d0 100644 --- a/src/kafka/msk_iam.rs +++ b/src/kafka/msk_iam.rs @@ -49,6 +49,8 @@ pub enum MskIamError { Credentials { source: aws_credential_types::provider::error::CredentialsError, }, + #[snafu(display("timed out after {timeout:?} loading AWS credentials for MSK IAM auth"))] + CredentialsTimeout { timeout: Duration }, #[snafu(display("failed to build SigV4 signing params: {source}"))] SigningParams { source: BoxError }, #[snafu(display("failed to presign MSK IAM token: {source}"))] @@ -72,14 +74,33 @@ pub struct MskAuthToken { /// component (and thus the last `CredentialsCache` clone) is dropped. pub(crate) async fn spawn_credentials_cache( provider: SharedCredentialsProvider, + load_timeout: Duration, ) -> Result { - let initial = provider.provide_credentials().await.context(CredentialsSnafu)?; + let initial = load_credentials(&provider, load_timeout).await?; let cache: CredentialsCache = Arc::new(Mutex::new(initial)); - tokio::spawn(refresh_loop(provider, Arc::downgrade(&cache))); + tokio::spawn(refresh_loop(provider, load_timeout, Arc::downgrade(&cache))); Ok(cache) } -async fn refresh_loop(provider: SharedCredentialsProvider, weak: Weak>) { +/// Resolve credentials, honoring the configured load timeout so a stalled IMDS/STS/profile lookup +/// can't hang the component build (matches `AwsAuthentication::credentials_cache` semantics). +async fn load_credentials( + provider: &SharedCredentialsProvider, + load_timeout: Duration, +) -> Result { + match tokio::time::timeout(load_timeout, provider.provide_credentials()).await { + Ok(res) => res.context(CredentialsSnafu), + Err(_) => Err(MskIamError::CredentialsTimeout { + timeout: load_timeout, + }), + } +} + +async fn refresh_loop( + provider: SharedCredentialsProvider, + load_timeout: Duration, + weak: Weak>, +) { loop { let sleep_for = match weak.upgrade() { None => break, @@ -88,8 +109,8 @@ async fn refresh_loop(provider: SharedCredentialsProvider, weak: Weak(&'a self) -> future::ProvideCredentials<'a> + where + Self: 'a, + { + // Never resolves within the timeout window. + future::ProvideCredentials::new(async { + tokio::time::sleep(Duration::from_secs(3600)).await; + unreachable!() + }) + } + } + + let provider = SharedCredentialsProvider::new(StalledProvider); + let err = spawn_credentials_cache(provider, Duration::from_secs(5)) + .await + .expect_err("must time out"); + assert!(matches!(err, MskIamError::CredentialsTimeout { .. }), "got: {err}"); + } + #[test] fn refresh_uses_skew_and_floor() { // Expiry far out -> refresh ~skew before it. diff --git a/src/sinks/kafka/config.rs b/src/sinks/kafka/config.rs index f5921d1e81feb..2849b1411e8f2 100644 --- a/src/sinks/kafka/config.rs +++ b/src/sinks/kafka/config.rs @@ -176,6 +176,12 @@ impl KafkaSinkConfig { self.auth.apply(&mut client_config)?; + // `librdkafka_options` are applied below, after `auth.apply`; reject any that would clobber + // the MSK IAM security/mechanism settings. + #[cfg(feature = "aws-core")] + self.auth + .validate_librdkafka_overrides(self.librdkafka_options.keys())?; + // All batch options are producer only. client_config .set("compression.codec", to_string(self.compression)) diff --git a/src/sources/kafka.rs b/src/sources/kafka.rs index 66dd89bd34534..1ecfba80b1e85 100644 --- a/src/sources/kafka.rs +++ b/src/sources/kafka.rs @@ -1231,6 +1231,12 @@ fn create_consumer( config.auth.apply(&mut client_config)?; if let Some(librdkafka_options) = &config.librdkafka_options { + // `librdkafka_options` override values set by `auth.apply`; reject any that would clobber the + // MSK IAM security/mechanism settings. + #[cfg(feature = "aws-core")] + config + .auth + .validate_librdkafka_overrides(librdkafka_options.keys())?; for (key, value) in librdkafka_options { client_config.set(key.as_str(), value.as_str()); } From 1cf671f04f2d3591339dccb6aef29b44af631e0d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Mon, 6 Jul 2026 20:12:35 +0200 Subject: [PATCH 06/10] fix(kafka): clamp MSK IAM token to credential expiry, partition-aware signing host, reject librdkafka OAUTHBEARER; add cue docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fourth round of review fixes: - Clamp the OAUTHBEARER token lifetime to the credential expiry: temporary (STS/IMDS) credentials can expire before the fixed 15-minute presign window, so advertise the earlier of (signing_time + 900s) and credentials.expiry() to force a refresh before the AWS session dies. - Partition-aware signing host: resolve kafka.{region}.amazonaws.com.cn for China regions (and honor an explicit sasl.aws_msk_iam.endpoint) instead of hard-coding the commercial suffix, which would fail SigV4 validation in other partitions. - Reject librdkafka_options.sasl.mechanism = OAUTHBEARER when MSK IAM is not enabled: the refresh callback is always-on in aws-core builds, so this closes the gap where OAUTHBEARER set via librdkafka_options (not the mechanism field) would reach it with no token provider. validate_librdkafka_overrides now inspects values, not just keys. - Regenerate Cue component docs for the new sasl.aws_msk_iam options (kafka source and sink). Add tests: signing host (partition + endpoint), lifetime clamp, and librdkafka OAUTHBEARER rejection. Signed-off-by: Gaál György --- src/kafka.rs | 81 ++++++--- src/kafka/msk_iam.rs | 108 ++++++++++-- src/sinks/kafka/config.rs | 2 +- src/sources/kafka.rs | 2 +- .../components/sinks/generated/kafka.cue | 154 +++++++++++++++++ .../components/sources/generated/kafka.cue | 158 +++++++++++++++++- 6 files changed, 463 insertions(+), 42 deletions(-) diff --git a/src/kafka.rs b/src/kafka.rs index bc6105c58bfb2..a261930fccbca 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -268,6 +268,9 @@ impl KafkaAuthConfig { let region = cfg.region.region().ok_or( "`sasl.aws_msk_iam` requires a region (set `sasl.aws_msk_iam.region`)", )?; + // The signer host honors an explicit endpoint and resolves the partition suffix (e.g. China + // regions use `amazonaws.com.cn`). + let host = msk_iam::signing_host(region.as_ref(), cfg.region.endpoint().as_deref()); let provider = cfg .auth .credentials_provider(region.clone(), proxy, None) @@ -276,17 +279,25 @@ impl KafkaAuthConfig { // refreshed in the background so the token callback stays synchronous. let credentials = msk_iam::spawn_credentials_cache(provider, cfg.auth.load_timeout()).await?; - Ok(Some(MskIamCredentials { region, credentials })) + Ok(Some(MskIamCredentials { + region, + host, + credentials, + })) } - /// Reject `librdkafka_options` that would override the security/mechanism settings MSK IAM - /// depends on. Both the source and sink apply `librdkafka_options` *after* [`Self::apply`], so a - /// leftover `security.protocol`/`sasl.*` override would silently drop OAUTHBEARER/SASL_SSL and - /// the component would load credentials but fail to authenticate. + /// Validate `librdkafka_options` against the MSK IAM / OAUTHBEARER constraints. Both the source + /// and sink apply `librdkafka_options` *after* [`Self::apply`], so: + /// + /// * When MSK IAM is enabled, a leftover `security.protocol`/`sasl.*` override would silently + /// drop OAUTHBEARER/SASL_SSL — the component would load credentials but fail to authenticate. + /// * When MSK IAM is not enabled, setting `sasl.mechanism = OAUTHBEARER` here would still reach + /// the always-on refresh callback (which has no token provider) — reject it with the same + /// guidance as the top-level `sasl.mechanism` guard. #[cfg(feature = "aws-core")] pub(crate) fn validate_librdkafka_overrides<'a>( &self, - option_keys: impl IntoIterator, + options: impl IntoIterator, ) -> crate::Result<()> { let msk_iam_enabled = self .sasl @@ -294,22 +305,30 @@ impl KafkaAuthConfig { .and_then(|s| s.aws_msk_iam.as_ref()) .map(|c| c.enabled) .unwrap_or(false); - if !msk_iam_enabled { - return Ok(()); - } + const RESERVED: [&str; 4] = [ "security.protocol", "sasl.mechanism", "sasl.username", "sasl.password", ]; - for key in option_keys { - if RESERVED.iter().any(|r| key.eq_ignore_ascii_case(r)) { - return Err(format!( - "`librdkafka_options.{key}` conflicts with `sasl.aws_msk_iam`; remove it \ - (MSK IAM manages `security.protocol` and `sasl.*`)" - ) - .into()); + for (key, value) in options { + if msk_iam_enabled { + if RESERVED.iter().any(|r| key.eq_ignore_ascii_case(r)) { + return Err(format!( + "`librdkafka_options.{key}` conflicts with `sasl.aws_msk_iam`; remove it \ + (MSK IAM manages `security.protocol` and `sasl.*`)" + ) + .into()); + } + } else if key.eq_ignore_ascii_case("sasl.mechanism") + && value.eq_ignore_ascii_case("OAUTHBEARER") + { + return Err( + "`librdkafka_options.sasl.mechanism = \"OAUTHBEARER\"` is only supported via \ + `sasl.aws_msk_iam` (AWS MSK IAM)" + .into(), + ); } } Ok(()) @@ -322,6 +341,8 @@ impl KafkaAuthConfig { #[derive(Clone)] pub(crate) struct MskIamCredentials { pub(crate) region: Region, + /// The SigV4 signing host (partition-aware, honoring an explicit endpoint). + pub(crate) host: String, pub(crate) credentials: msk_iam::CredentialsCache, } @@ -394,7 +415,8 @@ impl ClientContext for KafkaStatisticsContext { .lock() .map_err(|_| "MSK IAM credential cache mutex poisoned")? .clone(); - let token = msk_iam::build_token(&creds.region, &credentials, SystemTime::now())?; + let token = + msk_iam::build_token(&creds.region, &creds.host, &credentials, SystemTime::now())?; Ok(OAuthToken { token: token.token, @@ -607,26 +629,43 @@ mod auth_tests { ] { let opts = HashMap::from([(key.to_string(), "x".to_string())]); let err = auth - .validate_librdkafka_overrides(opts.keys()) + .validate_librdkafka_overrides(opts.iter()) .expect_err("conflicting override must be rejected"); assert!(err.to_string().contains(key), "got: {err}"); } // Benign options are fine. let ok = HashMap::from([("socket.timeout.ms".to_string(), "1000".to_string())]); - assert!(auth.validate_librdkafka_overrides(ok.keys()).is_ok()); + assert!(auth.validate_librdkafka_overrides(ok.iter()).is_ok()); } #[cfg(feature = "aws-core")] #[test] fn librdkafka_overrides_ignored_without_msk_iam() { - // Without MSK IAM these keys are the user's own business. + // Without MSK IAM, non-OAUTHBEARER overrides are the user's own business. use std::collections::HashMap; let auth = KafkaAuthConfig { sasl: Some(sasl(Some(true), Some("SCRAM-SHA-512"), Some("u"))), tls: Some(tls(true)), }; let opts = HashMap::from([("sasl.mechanism".to_string(), "PLAIN".to_string())]); - assert!(auth.validate_librdkafka_overrides(opts.keys()).is_ok()); + assert!(auth.validate_librdkafka_overrides(opts.iter()).is_ok()); + } + + #[cfg(feature = "aws-core")] + #[test] + fn librdkafka_oauthbearer_without_msk_iam_is_rejected() { + // Closes the gap where OAUTHBEARER set via librdkafka_options (not the mechanism field) would + // still reach the always-on refresh callback with no token provider. + use std::collections::HashMap; + let auth = KafkaAuthConfig { + sasl: Some(sasl(Some(true), None, None)), + tls: Some(tls(true)), + }; + let opts = HashMap::from([("sasl.mechanism".to_string(), "OAUTHBEARER".to_string())]); + let err = auth + .validate_librdkafka_overrides(opts.iter()) + .expect_err("librdkafka OAUTHBEARER without MSK IAM must be rejected"); + assert!(err.to_string().contains("aws_msk_iam"), "got: {err}"); } #[cfg(feature = "aws-core")] diff --git a/src/kafka/msk_iam.rs b/src/kafka/msk_iam.rs index 5acef1d5a61d0..32a70de480cba 100644 --- a/src/kafka/msk_iam.rs +++ b/src/kafka/msk_iam.rs @@ -129,35 +129,60 @@ fn next_refresh(expiry: Option) -> Duration { } } -/// Mint a fresh MSK IAM SASL/OAUTHBEARER token for `region` from already-resolved `credentials`. -/// Fully synchronous — safe to call from librdkafka's refresh callback on any thread. +/// The MSK IAM signing host for a region, honoring an explicit `endpoint` override and otherwise +/// resolving the partition suffix (e.g. `amazonaws.com.cn` for China regions). +pub(crate) fn signing_host(region: &str, endpoint: Option<&str>) -> String { + if let Some(endpoint) = endpoint { + // Strip scheme and any trailing slash so it can be used as a bare host. + return endpoint + .trim_start_matches("https://") + .trim_start_matches("http://") + .trim_end_matches('/') + .to_string(); + } + let suffix = if region.starts_with("cn-") { + "amazonaws.com.cn" + } else { + "amazonaws.com" + }; + format!("kafka.{region}.{suffix}") +} + +/// Mint a fresh MSK IAM SASL/OAUTHBEARER token for `region`/`host` from already-resolved +/// `credentials`. Fully synchronous — safe to call from librdkafka's refresh callback on any thread. pub(crate) fn build_token( region: &Region, + host: &str, credentials: &Credentials, signing_time: SystemTime, ) -> Result { - let signed_url = presign(region, credentials, signing_time)?; + let signed_url = presign(region, host, credentials, signing_time)?; // `User-Agent` is appended *after* signing (unsigned), matching the AWS signers. let signed_url = format!("{signed_url}&User-Agent={USER_AGENT}"); let token = URL_SAFE_NO_PAD.encode(signed_url.as_bytes()); - let signing_ms = signing_time + // The token can't outlive the credentials that signed it: a presign is valid for 15 min, but + // temporary (STS/IMDS) credentials may expire sooner. Advertise the earlier of the two so + // librdkafka refreshes before the underlying AWS session dies. + let token_expiry = signing_time + Duration::from_secs(EXPIRY_SECONDS); + let effective_expiry = match credentials.expiry() { + Some(cred_expiry) if cred_expiry < token_expiry => cred_expiry, + _ => token_expiry, + }; + let lifetime_ms = effective_expiry .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_millis() as i64; - Ok(MskAuthToken { - token, - lifetime_ms: signing_ms + (EXPIRY_SECONDS as i64) * 1000, - }) + Ok(MskAuthToken { token, lifetime_ms }) } -/// Build the SigV4-presigned `https://kafka.{region}.amazonaws.com/?Action=...` URL. +/// Build the SigV4-presigned `https://{host}/?Action=...` URL. fn presign( region: &Region, + host: &str, credentials: &Credentials, time: SystemTime, ) -> Result { - let host = format!("kafka.{}.amazonaws.com", region.as_ref()); let identity: Identity = credentials.clone().into(); let mut settings = SigningSettings::default(); @@ -174,7 +199,7 @@ fn presign( .map_err(|e| MskIamError::SigningParams { source: Box::new(e) })?; let url = format!("https://{host}/?Action={}", rfc3986(ACTION)); - let headers = [("host", host.as_str())]; + let headers = [("host", host)]; let signable = SignableRequest::new("GET", &url, headers.into_iter(), SignableBody::Bytes(&[])) .map_err(|e| MskIamError::Signing { source: Box::new(e) })?; @@ -225,6 +250,9 @@ mod tests { const GOLDEN_SIG_NO_TOKEN: &str = "12f697bba1fe524b7e9f370e2c56075ea612d9d4d2c7f16c402ffc72520fce5e"; + // The oracle signed `kafka.ap-southeast-1.amazonaws.com` for this region. + const HOST: &str = "kafka.ap-southeast-1.amazonaws.com"; + fn region() -> Region { Region::new("ap-southeast-1") } @@ -271,14 +299,14 @@ mod tests { /// matches the independent oracle — order-independent because MSK re-canonicalizes the query. #[test] fn full_presigned_url_matches_oracle() { - let url = presign(®ion(), &creds(true), fixed_time()).expect("presign"); + let url = presign(®ion(), HOST, &creds(true), fixed_time()).expect("presign"); assert_eq!(normalize(&url), normalize(ORACLE_URL_WITH_TOKEN), "url={url}"); } /// KNOWN-ANSWER: aws-sigv4 signature must equal the oracle's, with a session token present. #[test] fn signature_matches_oracle_with_session_token() { - let url = presign(®ion(), &creds(true), fixed_time()).expect("presign"); + let url = presign(®ion(), HOST, &creds(true), fixed_time()).expect("presign"); assert_eq!(query_param(&url, "X-Amz-Signature"), Some(GOLDEN_SIG_WITH_TOKEN)); assert_eq!(query_param(&url, "X-Amz-Security-Token"), Some("IQoJb3Jp")); } @@ -286,7 +314,7 @@ mod tests { /// KNOWN-ANSWER: same, without a session token (static long-term credentials). #[test] fn signature_matches_oracle_without_session_token() { - let url = presign(®ion(), &creds(false), fixed_time()).expect("presign"); + let url = presign(®ion(), HOST, &creds(false), fixed_time()).expect("presign"); assert_eq!(query_param(&url, "X-Amz-Signature"), Some(GOLDEN_SIG_NO_TOKEN)); assert_eq!(query_param(&url, "X-Amz-Security-Token"), None); } @@ -294,7 +322,7 @@ mod tests { /// Structural invariants the MSK broker requires of the presigned URL. #[test] fn presigned_url_has_required_msk_shape() { - let url = presign(®ion(), &creds(false), fixed_time()).expect("presign"); + let url = presign(®ion(), HOST, &creds(false), fixed_time()).expect("presign"); assert!(url.starts_with("https://kafka.ap-southeast-1.amazonaws.com/?")); assert_eq!(query_param(&url, "Action"), Some("kafka-cluster%3AConnect")); assert_eq!(query_param(&url, "X-Amz-Algorithm"), Some("AWS4-HMAC-SHA256")); @@ -310,7 +338,7 @@ mod tests { /// is the absolute unix-epoch ms of signing-time + 900s. #[test] fn build_token_encodes_and_sets_lifetime() { - let out = build_token(®ion(), &creds(false), fixed_time()).expect("token"); + let out = build_token(®ion(), HOST, &creds(false), fixed_time()).expect("token"); assert!(!out.token.contains('='), "token must be unpadded base64url"); let decoded = URL_SAFE_NO_PAD.decode(out.token.as_bytes()).expect("decode"); let url = String::from_utf8(decoded).expect("utf8"); @@ -320,6 +348,52 @@ mod tests { assert_eq!(out.lifetime_ms, (FIXED_EPOCH_SECS as i64) * 1000 + 900_000); } + #[test] + fn signing_host_resolves_partition_and_endpoint() { + // Commercial partition. + assert_eq!( + signing_host("us-east-1", None), + "kafka.us-east-1.amazonaws.com" + ); + // China partition uses the .com.cn suffix. + assert_eq!( + signing_host("cn-north-1", None), + "kafka.cn-north-1.amazonaws.com.cn" + ); + // Explicit endpoint overrides, with scheme/trailing slash stripped. + assert_eq!( + signing_host("us-east-1", Some("https://kafka.example.internal/")), + "kafka.example.internal" + ); + } + + /// A token must never be advertised as valid past the credentials that signed it. + #[test] + fn lifetime_clamped_to_credential_expiry() { + // Credentials expiring 300s after signing time — earlier than the 900s presign window. + let cred_expiry = fixed_time() + Duration::from_secs(300); + let creds = Credentials::new( + ACCESS_KEY, + SECRET_KEY, + Some(SESSION_TOKEN.to_string()), + Some(cred_expiry), + "test", + ); + let out = build_token(®ion(), HOST, &creds, fixed_time()).expect("token"); + assert_eq!(out.lifetime_ms, (FIXED_EPOCH_SECS as i64) * 1000 + 300_000); + + // Credentials outliving the window -> full 900s. + let long = Credentials::new( + ACCESS_KEY, + SECRET_KEY, + Some(SESSION_TOKEN.to_string()), + Some(fixed_time() + Duration::from_secs(3600)), + "test", + ); + let out = build_token(®ion(), HOST, &long, fixed_time()).expect("token"); + assert_eq!(out.lifetime_ms, (FIXED_EPOCH_SECS as i64) * 1000 + 900_000); + } + /// The cache path: resolve via a static provider, then mint a token from the cached credentials. #[tokio::test] async fn credentials_cache_then_build_token() { @@ -328,7 +402,7 @@ mod tests { .await .expect("cache"); let cached = cache.lock().unwrap().clone(); - let out = build_token(®ion(), &cached, fixed_time()).expect("token"); + let out = build_token(®ion(), HOST, &cached, fixed_time()).expect("token"); let decoded = URL_SAFE_NO_PAD.decode(out.token.as_bytes()).unwrap(); let url = String::from_utf8(decoded).unwrap(); assert!(url.contains(GOLDEN_SIG_WITH_TOKEN)); diff --git a/src/sinks/kafka/config.rs b/src/sinks/kafka/config.rs index 2849b1411e8f2..c44c67d114250 100644 --- a/src/sinks/kafka/config.rs +++ b/src/sinks/kafka/config.rs @@ -180,7 +180,7 @@ impl KafkaSinkConfig { // the MSK IAM security/mechanism settings. #[cfg(feature = "aws-core")] self.auth - .validate_librdkafka_overrides(self.librdkafka_options.keys())?; + .validate_librdkafka_overrides(self.librdkafka_options.iter())?; // All batch options are producer only. client_config diff --git a/src/sources/kafka.rs b/src/sources/kafka.rs index 1ecfba80b1e85..041df3ae3f3a1 100644 --- a/src/sources/kafka.rs +++ b/src/sources/kafka.rs @@ -1236,7 +1236,7 @@ fn create_consumer( #[cfg(feature = "aws-core")] config .auth - .validate_librdkafka_overrides(librdkafka_options.keys())?; + .validate_librdkafka_overrides(librdkafka_options.iter())?; for (key, value) in librdkafka_options { client_config.set(key.as_str(), value.as_str()); } diff --git a/website/cue/reference/components/sinks/generated/kafka.cue b/website/cue/reference/components/sinks/generated/kafka.cue index e8b75370974bd..8ed823bfc0cc9 100644 --- a/website/cue/reference/components/sinks/generated/kafka.cue +++ b/website/cue/reference/components/sinks/generated/kafka.cue @@ -600,6 +600,160 @@ generated: components: sinks: kafka: configuration: { description: "Configuration for SASL authentication when interacting with Kafka." required: false type: object: options: { + aws_msk_iam: { + description: """ + AWS MSK IAM authentication using SASL/OAUTHBEARER. + + When enabled, `mechanism` is forced to `OAUTHBEARER` and the token is generated by + SigV4-presigning the `kafka-cluster:Connect` action against the configured region. Mutually + exclusive with `username`/`password`. Requires TLS (MSK IAM listens on the SASL_SSL port, + `9098`). + """ + required: false + type: object: options: { + auth: { + description: "Configuration of the authentication strategy for interacting with AWS services." + required: false + type: object: options: { + access_key_id: { + description: "The AWS access key ID." + required: true + type: string: examples: ["AKIAIOSFODNN7EXAMPLE"] + } + assume_role: { + description: """ + The ARN of an [IAM role][iam_role] to assume. + + [iam_role]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html + """ + required: true + type: string: examples: ["arn:aws:iam::123456789098:role/my_role"] + } + credentials_file: { + description: "Path to the credentials file." + required: true + type: string: examples: ["/my/aws/credentials"] + } + external_id: { + description: """ + The optional unique external ID in conjunction with role to assume. + + [external_id]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html + """ + required: false + type: string: examples: ["randomEXAMPLEidString"] + } + imds: { + description: "Configuration for authenticating with AWS through IMDS." + required: false + type: object: options: { + connect_timeout_seconds: { + description: "Connect timeout for IMDS." + required: false + type: uint: { + default: 1 + unit: "seconds" + } + } + max_attempts: { + description: "Number of IMDS retries for fetching tokens and metadata." + required: false + type: uint: default: 4 + } + read_timeout_seconds: { + description: "Read timeout for IMDS." + required: false + type: uint: { + default: 1 + unit: "seconds" + } + } + } + } + load_timeout_secs: { + description: """ + Timeout for successfully loading any credentials, in seconds. + + Relevant when the default credentials chain or `assume_role` is used. + """ + required: false + type: uint: { + examples: [30] + unit: "seconds" + } + } + profile: { + description: """ + The credentials profile to use. + + Used to select AWS credentials from a provided credentials file. + """ + required: false + type: string: { + default: "default" + examples: ["develop"] + } + } + region: { + description: """ + The [AWS region][aws_region] to send STS requests to. + + If not set, this defaults to the configured region + for the service itself. + + [aws_region]: https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints + """ + required: false + type: string: examples: ["us-west-2"] + } + secret_access_key: { + description: "The AWS secret access key." + required: true + type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"] + } + session_name: { + description: """ + The optional [RoleSessionName][role_session_name] is a unique session identifier for your assumed role. + + Should be unique per principal or reason. + If not set, the session name is autogenerated like assume-role-provider-1736428351340 + + [role_session_name]: https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html + """ + required: false + type: string: examples: ["vector-indexer-role"] + } + session_token: { + description: """ + The AWS session token. + See [AWS temporary credentials](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_use-resources.html) + """ + required: false + type: string: examples: ["AQoDYXdz...AQoDYXdz..."] + } + } + } + enabled: { + description: "Enables MSK IAM authentication." + required: true + type: bool: {} + } + endpoint: { + description: "Custom endpoint for use with AWS-compatible services." + required: false + type: string: examples: ["http://127.0.0.0:5000/path/to/service"] + } + region: { + description: """ + The [AWS region][aws_region] of the target service. + + [aws_region]: https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints + """ + required: false + type: string: examples: ["us-east-1"] + } + } + } enabled: { description: """ Enables SASL authentication. diff --git a/website/cue/reference/components/sources/generated/kafka.cue b/website/cue/reference/components/sources/generated/kafka.cue index 0979aa41c30a0..945d5cf70b71a 100644 --- a/website/cue/reference/components/sources/generated/kafka.cue +++ b/website/cue/reference/components/sources/generated/kafka.cue @@ -653,7 +653,7 @@ generated: components: sources: kafka: configuration: { type: string: { default: "offset" examples: [ - "offset", + "offset" ] } } @@ -675,6 +675,160 @@ generated: components: sources: kafka: configuration: { description: "Configuration for SASL authentication when interacting with Kafka." required: false type: object: options: { + aws_msk_iam: { + description: """ + AWS MSK IAM authentication using SASL/OAUTHBEARER. + + When enabled, `mechanism` is forced to `OAUTHBEARER` and the token is generated by + SigV4-presigning the `kafka-cluster:Connect` action against the configured region. Mutually + exclusive with `username`/`password`. Requires TLS (MSK IAM listens on the SASL_SSL port, + `9098`). + """ + required: false + type: object: options: { + auth: { + description: "Configuration of the authentication strategy for interacting with AWS services." + required: false + type: object: options: { + access_key_id: { + description: "The AWS access key ID." + required: true + type: string: examples: ["AKIAIOSFODNN7EXAMPLE"] + } + assume_role: { + description: """ + The ARN of an [IAM role][iam_role] to assume. + + [iam_role]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html + """ + required: true + type: string: examples: ["arn:aws:iam::123456789098:role/my_role"] + } + credentials_file: { + description: "Path to the credentials file." + required: true + type: string: examples: ["/my/aws/credentials"] + } + external_id: { + description: """ + The optional unique external ID in conjunction with role to assume. + + [external_id]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html + """ + required: false + type: string: examples: ["randomEXAMPLEidString"] + } + imds: { + description: "Configuration for authenticating with AWS through IMDS." + required: false + type: object: options: { + connect_timeout_seconds: { + description: "Connect timeout for IMDS." + required: false + type: uint: { + default: 1 + unit: "seconds" + } + } + max_attempts: { + description: "Number of IMDS retries for fetching tokens and metadata." + required: false + type: uint: default: 4 + } + read_timeout_seconds: { + description: "Read timeout for IMDS." + required: false + type: uint: { + default: 1 + unit: "seconds" + } + } + } + } + load_timeout_secs: { + description: """ + Timeout for successfully loading any credentials, in seconds. + + Relevant when the default credentials chain or `assume_role` is used. + """ + required: false + type: uint: { + examples: [30] + unit: "seconds" + } + } + profile: { + description: """ + The credentials profile to use. + + Used to select AWS credentials from a provided credentials file. + """ + required: false + type: string: { + default: "default" + examples: ["develop"] + } + } + region: { + description: """ + The [AWS region][aws_region] to send STS requests to. + + If not set, this defaults to the configured region + for the service itself. + + [aws_region]: https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints + """ + required: false + type: string: examples: ["us-west-2"] + } + secret_access_key: { + description: "The AWS secret access key." + required: true + type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"] + } + session_name: { + description: """ + The optional [RoleSessionName][role_session_name] is a unique session identifier for your assumed role. + + Should be unique per principal or reason. + If not set, the session name is autogenerated like assume-role-provider-1736428351340 + + [role_session_name]: https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html + """ + required: false + type: string: examples: ["vector-indexer-role"] + } + session_token: { + description: """ + The AWS session token. + See [AWS temporary credentials](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_use-resources.html) + """ + required: false + type: string: examples: ["AQoDYXdz...AQoDYXdz..."] + } + } + } + enabled: { + description: "Enables MSK IAM authentication." + required: true + type: bool: {} + } + endpoint: { + description: "Custom endpoint for use with AWS-compatible services." + required: false + type: string: examples: ["http://127.0.0.0:5000/path/to/service"] + } + region: { + description: """ + The [AWS region][aws_region] of the target service. + + [aws_region]: https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints + """ + required: false + type: string: examples: ["us-east-1"] + } + } + } enabled: { description: """ Enables SASL authentication. @@ -843,7 +997,7 @@ generated: components: sources: kafka: configuration: { type: string: { default: "topic" examples: [ - "topic", + "topic" ] } } From d463e33e2dbfa6e2c9db2bf9d081b4792f37ee1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Mon, 6 Jul 2026 20:29:57 +0200 Subject: [PATCH 07/10] fix(kafka): also reject the plural sasl.mechanisms alias in librdkafka_options MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit librdkafka accepts both sasl.mechanism (singular) and sasl.mechanisms (plural) for the same property. validate_librdkafka_overrides only matched the singular form, so a leftover librdkafka_options.sasl.mechanisms could override MSK IAM's OAUTHBEARER (when enabled) or bypass the non-MSK OAUTHBEARER rejection. Match both aliases; extend tests. Signed-off-by: Gaál György --- src/kafka.rs | 41 ++++++++++++++++++++++------------------- 1 file changed, 22 insertions(+), 19 deletions(-) diff --git a/src/kafka.rs b/src/kafka.rs index a261930fccbca..30e921e8998a8 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -306,29 +306,29 @@ impl KafkaAuthConfig { .map(|c| c.enabled) .unwrap_or(false); - const RESERVED: [&str; 4] = [ - "security.protocol", - "sasl.mechanism", - "sasl.username", - "sasl.password", - ]; + // Non-mechanism keys that MSK IAM owns. `security.protocol` has no alias; `sasl.username` / + // `sasl.password` are canonical. + const RESERVED: [&str; 3] = ["security.protocol", "sasl.username", "sasl.password"]; + // librdkafka accepts both `sasl.mechanism` (singular) and `sasl.mechanisms` (plural) for the + // same property, so both must be caught. + let is_mechanism = + |k: &str| k.eq_ignore_ascii_case("sasl.mechanism") || k.eq_ignore_ascii_case("sasl.mechanisms"); + for (key, value) in options { if msk_iam_enabled { - if RESERVED.iter().any(|r| key.eq_ignore_ascii_case(r)) { + if is_mechanism(key) || RESERVED.iter().any(|r| key.eq_ignore_ascii_case(r)) { return Err(format!( "`librdkafka_options.{key}` conflicts with `sasl.aws_msk_iam`; remove it \ (MSK IAM manages `security.protocol` and `sasl.*`)" ) .into()); } - } else if key.eq_ignore_ascii_case("sasl.mechanism") - && value.eq_ignore_ascii_case("OAUTHBEARER") - { - return Err( - "`librdkafka_options.sasl.mechanism = \"OAUTHBEARER\"` is only supported via \ + } else if is_mechanism(key) && value.eq_ignore_ascii_case("OAUTHBEARER") { + return Err(format!( + "`librdkafka_options.{key} = \"OAUTHBEARER\"` is only supported via \ `sasl.aws_msk_iam` (AWS MSK IAM)" - .into(), - ); + ) + .into()); } } Ok(()) @@ -624,6 +624,7 @@ mod auth_tests { for key in [ "security.protocol", "sasl.mechanism", + "sasl.mechanisms", // librdkafka plural alias "sasl.username", "sasl.password", ] { @@ -661,11 +662,13 @@ mod auth_tests { sasl: Some(sasl(Some(true), None, None)), tls: Some(tls(true)), }; - let opts = HashMap::from([("sasl.mechanism".to_string(), "OAUTHBEARER".to_string())]); - let err = auth - .validate_librdkafka_overrides(opts.iter()) - .expect_err("librdkafka OAUTHBEARER without MSK IAM must be rejected"); - assert!(err.to_string().contains("aws_msk_iam"), "got: {err}"); + for key in ["sasl.mechanism", "sasl.mechanisms"] { + let opts = HashMap::from([(key.to_string(), "OAUTHBEARER".to_string())]); + let err = auth + .validate_librdkafka_overrides(opts.iter()) + .expect_err("librdkafka OAUTHBEARER without MSK IAM must be rejected"); + assert!(err.to_string().contains("aws_msk_iam"), "got: {err}"); + } } #[cfg(feature = "aws-core")] From 9d269acb650fc6cad14450917d52b0a896cb66d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Mon, 6 Jul 2026 22:13:27 +0200 Subject: [PATCH 08/10] fix(kafka): strip endpoint path, validate before credential load, reject MSK IAM without aws-core MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fifth round of review fixes: - signing_host now reduces an explicit endpoint to the bare authority (host[:port]), dropping any /path, ?query, or #fragment — a documented full-URL endpoint would otherwise be signed as an invalid Host and fail SigV4 validation. - Validate the client/auth config (TLS requirement, conflicting librdkafka_options, unsupported-feature) before resolving AWS credentials in both the source and sink build paths, so an invalid config fails fast with an actionable error instead of blocking on IMDS/STS and reporting a credential error. - The aws_msk_iam block now parses even without the aws-core feature (its AWS-typed fields are cfg-gated), and apply() rejects an *enabled* MSK IAM config in a non-aws-core build instead of silently connecting without OAUTHBEARER/SASL_SSL. Regenerate Cue docs; add tests for endpoint-path stripping and the aws-core rejection. Signed-off-by: Gaál György --- src/kafka.rs | 38 +++++++++++++++++-- src/kafka/msk_iam.rs | 16 +++++--- src/sinks/kafka/config.rs | 5 +++ src/sources/kafka.rs | 12 ++++++ .../components/sinks/generated/kafka.cue | 3 ++ .../components/sources/generated/kafka.cue | 3 ++ 6 files changed, 68 insertions(+), 9 deletions(-) diff --git a/src/kafka.rs b/src/kafka.rs index 30e921e8998a8..20a29997364d8 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -102,23 +102,26 @@ pub struct KafkaSaslConfig { /// SigV4-presigning the `kafka-cluster:Connect` action against the configured region. Mutually /// exclusive with `username`/`password`. Requires TLS (MSK IAM listens on the SASL_SSL port, /// `9098`). - #[cfg(feature = "aws-core")] + /// + /// Requires Vector to be built with the `aws-core` feature; otherwise enabling it is a + /// configuration error rather than being silently ignored. #[configurable(derived)] pub(crate) aws_msk_iam: Option, } /// Configuration for AWS MSK IAM authentication (SASL/OAUTHBEARER). -#[cfg(feature = "aws-core")] #[configurable_component] #[derive(Clone, Debug, Default)] pub struct AwsMskIamConfig { /// Enables MSK IAM authentication. pub(crate) enabled: bool, + #[cfg(feature = "aws-core")] #[configurable(derived)] #[serde(flatten, default)] pub(crate) region: RegionOrEndpoint, + #[cfg(feature = "aws-core")] #[configurable(derived)] #[serde(default)] pub(crate) auth: AwsAuthentication, @@ -126,15 +129,22 @@ pub struct AwsMskIamConfig { impl KafkaAuthConfig { pub(crate) fn apply(&self, client: &mut ClientConfig) -> crate::Result<()> { - #[cfg(feature = "aws-core")] let msk_iam_enabled = self .sasl .as_ref() .and_then(|s| s.aws_msk_iam.as_ref()) .map(|c| c.enabled) .unwrap_or(false); + + // The `aws_msk_iam` config parses even without the `aws-core` feature (the block is inert), + // so reject an *enabled* MSK IAM config in that build instead of silently connecting without + // OAUTHBEARER/SASL_SSL. #[cfg(not(feature = "aws-core"))] - let msk_iam_enabled = false; + if msk_iam_enabled { + return Err( + "`sasl.aws_msk_iam` requires Vector to be built with the `aws-core` feature".into(), + ); + } // MSK IAM (SASL/OAUTHBEARER) implies SASL even if the legacy `sasl.enabled` flag is unset. let sasl_enabled = @@ -671,6 +681,26 @@ mod auth_tests { } } + #[cfg(not(feature = "aws-core"))] + #[test] + fn msk_iam_without_aws_core_feature_is_rejected() { + // The `aws_msk_iam` block still parses without `aws-core`, but enabling it must be a hard + // error rather than being silently ignored (which would connect without OAUTHBEARER). + let auth = KafkaAuthConfig { + sasl: Some(KafkaSaslConfig { + enabled: Some(true), + aws_msk_iam: Some(AwsMskIamConfig { enabled: true }), + ..Default::default() + }), + tls: Some(tls(true)), + }; + let mut cc = ClientConfig::new(); + let err = auth + .apply(&mut cc) + .expect_err("must reject MSK IAM without the aws-core feature"); + assert!(err.to_string().contains("aws-core"), "got: {err}"); + } + #[cfg(feature = "aws-core")] #[test] fn msk_iam_without_tls_is_rejected() { diff --git a/src/kafka/msk_iam.rs b/src/kafka/msk_iam.rs index 32a70de480cba..e8bc4a200f566 100644 --- a/src/kafka/msk_iam.rs +++ b/src/kafka/msk_iam.rs @@ -133,12 +133,13 @@ fn next_refresh(expiry: Option) -> Duration { /// resolving the partition suffix (e.g. `amazonaws.com.cn` for China regions). pub(crate) fn signing_host(region: &str, endpoint: Option<&str>) -> String { if let Some(endpoint) = endpoint { - // Strip scheme and any trailing slash so it can be used as a bare host. - return endpoint + // Reduce to the bare authority (`host[:port]`): strip the scheme, then drop any + // `/path`, `?query`, or `#fragment` so it is valid as both the URL host and Host header. + let no_scheme = endpoint .trim_start_matches("https://") - .trim_start_matches("http://") - .trim_end_matches('/') - .to_string(); + .trim_start_matches("http://"); + let authority = no_scheme.split(['/', '?', '#']).next().unwrap_or(no_scheme); + return authority.to_string(); } let suffix = if region.starts_with("cn-") { "amazonaws.com.cn" @@ -365,6 +366,11 @@ mod tests { signing_host("us-east-1", Some("https://kafka.example.internal/")), "kafka.example.internal" ); + // A documented full-URL endpoint with a path reduces to just the authority. + assert_eq!( + signing_host("us-east-1", Some("http://127.0.0.1:5000/path/to/service")), + "127.0.0.1:5000" + ); } /// A token must never be advertised as valid past the credentials that signed it. diff --git a/src/sinks/kafka/config.rs b/src/sinks/kafka/config.rs index c44c67d114250..a64903623e1f7 100644 --- a/src/sinks/kafka/config.rs +++ b/src/sinks/kafka/config.rs @@ -286,6 +286,11 @@ impl GenerateConfig for KafkaSinkConfig { #[typetag::serde(name = "kafka")] impl SinkConfig for KafkaSinkConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { + // Validate the client/auth config (TLS requirement, conflicting `librdkafka_options`, + // unsupported-feature) up front, so a misconfiguration fails fast rather than after the + // potentially-slow MSK IAM credential load below. + let _ = self.to_rdkafka()?; + #[cfg(feature = "aws-core")] let msk_iam = self.auth.msk_iam_credentials(&cx.proxy).await?; diff --git a/src/sources/kafka.rs b/src/sources/kafka.rs index 041df3ae3f3a1..0905cef8b4722 100644 --- a/src/sources/kafka.rs +++ b/src/sources/kafka.rs @@ -342,6 +342,18 @@ impl SourceConfig for KafkaSourceConfig { ); } + // Validate the client/auth config (TLS requirement, conflicting `librdkafka_options`, + // unsupported-feature) up front, so a misconfiguration fails fast rather than after the + // potentially-slow MSK IAM credential load below. + { + let mut probe = ClientConfig::new(); + self.auth.apply(&mut probe)?; + #[cfg(feature = "aws-core")] + if let Some(options) = &self.librdkafka_options { + self.auth.validate_librdkafka_overrides(options.iter())?; + } + } + #[cfg(feature = "aws-core")] let msk_iam = self.auth.msk_iam_credentials(&cx.proxy).await?; diff --git a/website/cue/reference/components/sinks/generated/kafka.cue b/website/cue/reference/components/sinks/generated/kafka.cue index 8ed823bfc0cc9..12d13ce3c0cb7 100644 --- a/website/cue/reference/components/sinks/generated/kafka.cue +++ b/website/cue/reference/components/sinks/generated/kafka.cue @@ -608,6 +608,9 @@ generated: components: sinks: kafka: configuration: { SigV4-presigning the `kafka-cluster:Connect` action against the configured region. Mutually exclusive with `username`/`password`. Requires TLS (MSK IAM listens on the SASL_SSL port, `9098`). + + Requires Vector to be built with the `aws-core` feature; otherwise enabling it is a + configuration error rather than being silently ignored. """ required: false type: object: options: { diff --git a/website/cue/reference/components/sources/generated/kafka.cue b/website/cue/reference/components/sources/generated/kafka.cue index 945d5cf70b71a..296960f0a154a 100644 --- a/website/cue/reference/components/sources/generated/kafka.cue +++ b/website/cue/reference/components/sources/generated/kafka.cue @@ -683,6 +683,9 @@ generated: components: sources: kafka: configuration: { SigV4-presigning the `kafka-cluster:Connect` action against the configured region. Mutually exclusive with `username`/`password`. Requires TLS (MSK IAM listens on the SASL_SSL port, `9098`). + + Requires Vector to be built with the `aws-core` feature; otherwise enabling it is a + configuration error rather than being silently ignored. """ required: false type: object: options: { From 249a686e07bd56205f3d3b000856f0ad50355431 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Mon, 6 Jul 2026 22:57:40 +0200 Subject: [PATCH 09/10] fix(kafka): stop rejecting non-MSK OAUTHBEARER (preserve librdkafka OIDC) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previous rounds proactively rejected a non-MSK OAUTHBEARER mechanism (top-level and via librdkafka_options). That was too aggressive: librdkafka has a built-in OIDC flow (sasl.oauthbearer.method=oidc) that does not use the application refresh callback, so those configs work fine even with ENABLE_REFRESH_OAUTH_TOKEN on (verified: method=oidc never invokes the app callback). Rejecting them broke a legitimate, historically-supported path. Remove both rejections. validate_librdkafka_overrides now only guards the keys MSK IAM manages when MSK IAM is actually enabled, and passes everything through otherwise. A non-MSK method=default OAUTHBEARER with no token provider still fails, but clearly at connect time rather than being pre-emptively rejected. Update tests accordingly. Signed-off-by: Gaál György --- src/kafka.rs | 120 ++++++++++++++++++++------------------------------- 1 file changed, 47 insertions(+), 73 deletions(-) diff --git a/src/kafka.rs b/src/kafka.rs index 20a29997364d8..32b7f80e597ac 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -161,25 +161,11 @@ impl KafkaAuthConfig { ); } - // rust-rdkafka gates the OAUTHBEARER refresh callback on a compile-time associated const, so - // in `aws-core` builds it is enabled for the context whenever it exists. Vector only provides - // a token for it via MSK IAM. Reject a bare `OAUTHBEARER` mechanism up front with guidance, - // rather than letting the callback fire later with no token provider. - #[cfg(feature = "aws-core")] - if sasl_enabled - && !msk_iam_enabled - && self - .sasl - .as_ref() - .and_then(|s| s.mechanism.as_deref()) - .is_some_and(|m| m.eq_ignore_ascii_case("OAUTHBEARER")) - { - return Err( - "`sasl.mechanism = \"OAUTHBEARER\"` is only supported via `sasl.aws_msk_iam` \ - (AWS MSK IAM)" - .into(), - ); - } + // NOTE: we intentionally do *not* reject a non-MSK `OAUTHBEARER` mechanism. librdkafka has a + // built-in OIDC flow (`sasl.oauthbearer.method=oidc`) that does not use the app refresh + // callback, so those configs keep working even though `ENABLE_REFRESH_OAUTH_TOKEN` is on. + // Only when the app callback is actually invoked without MSK IAM (i.e. `method=default` and + // no `aws_msk_iam`) does it error — and that surfaces clearly at connect time. let protocol = match (sasl_enabled, tls_enabled) { (false, false) => "plaintext", @@ -296,14 +282,13 @@ impl KafkaAuthConfig { })) } - /// Validate `librdkafka_options` against the MSK IAM / OAUTHBEARER constraints. Both the source - /// and sink apply `librdkafka_options` *after* [`Self::apply`], so: + /// When MSK IAM is enabled, reject `librdkafka_options` that would override the security/mechanism + /// settings it manages. Both the source and sink apply `librdkafka_options` *after* [`Self::apply`], + /// so a leftover `security.protocol`/`sasl.*` override would silently drop OAUTHBEARER/SASL_SSL and + /// the component would load credentials but fail to authenticate. /// - /// * When MSK IAM is enabled, a leftover `security.protocol`/`sasl.*` override would silently - /// drop OAUTHBEARER/SASL_SSL — the component would load credentials but fail to authenticate. - /// * When MSK IAM is not enabled, setting `sasl.mechanism = OAUTHBEARER` here would still reach - /// the always-on refresh callback (which has no token provider) — reject it with the same - /// guidance as the top-level `sasl.mechanism` guard. + /// This does nothing when MSK IAM is not enabled: non-MSK OAUTHBEARER (e.g. librdkafka's built-in + /// OIDC via `sasl.oauthbearer.method=oidc`) is a legitimate configuration and must pass through. #[cfg(feature = "aws-core")] pub(crate) fn validate_librdkafka_overrides<'a>( &self, @@ -315,28 +300,24 @@ impl KafkaAuthConfig { .and_then(|s| s.aws_msk_iam.as_ref()) .map(|c| c.enabled) .unwrap_or(false); + if !msk_iam_enabled { + return Ok(()); + } - // Non-mechanism keys that MSK IAM owns. `security.protocol` has no alias; `sasl.username` / - // `sasl.password` are canonical. - const RESERVED: [&str; 3] = ["security.protocol", "sasl.username", "sasl.password"]; - // librdkafka accepts both `sasl.mechanism` (singular) and `sasl.mechanisms` (plural) for the - // same property, so both must be caught. - let is_mechanism = - |k: &str| k.eq_ignore_ascii_case("sasl.mechanism") || k.eq_ignore_ascii_case("sasl.mechanisms"); - - for (key, value) in options { - if msk_iam_enabled { - if is_mechanism(key) || RESERVED.iter().any(|r| key.eq_ignore_ascii_case(r)) { - return Err(format!( - "`librdkafka_options.{key}` conflicts with `sasl.aws_msk_iam`; remove it \ - (MSK IAM manages `security.protocol` and `sasl.*`)" - ) - .into()); - } - } else if is_mechanism(key) && value.eq_ignore_ascii_case("OAUTHBEARER") { + // Keys MSK IAM owns. `security.protocol` has no alias; `sasl.username`/`sasl.password` are + // canonical; librdkafka accepts both `sasl.mechanism` and the plural `sasl.mechanisms`. + const RESERVED: [&str; 5] = [ + "security.protocol", + "sasl.username", + "sasl.password", + "sasl.mechanism", + "sasl.mechanisms", + ]; + for (key, _value) in options { + if RESERVED.iter().any(|r| key.eq_ignore_ascii_case(r)) { return Err(format!( - "`librdkafka_options.{key} = \"OAUTHBEARER\"` is only supported via \ - `sasl.aws_msk_iam` (AWS MSK IAM)" + "`librdkafka_options.{key}` conflicts with `sasl.aws_msk_iam`; remove it \ + (MSK IAM manages `security.protocol` and `sasl.*`)" ) .into()); } @@ -598,15 +579,16 @@ mod auth_tests { #[cfg(feature = "aws-core")] #[test] - fn bare_oauthbearer_mechanism_is_rejected() { - // OAUTHBEARER is only wired up for MSK IAM; a manual mechanism must fail fast with guidance. + fn non_msk_oauthbearer_mechanism_is_allowed() { + // A non-MSK OAUTHBEARER mechanism must NOT be rejected at config time: librdkafka's built-in + // OIDC flow doesn't use the app callback. `apply` sets it as a normal mechanism. let auth = KafkaAuthConfig { sasl: Some(sasl(Some(true), Some("OAUTHBEARER"), None)), tls: Some(tls(true)), }; - let mut cc = ClientConfig::new(); - let err = auth.apply(&mut cc).expect_err("bare OAUTHBEARER must be rejected"); - assert!(err.to_string().contains("aws_msk_iam"), "got: {err}"); + let cc = applied(&auth); + assert_eq!(cc.get("security.protocol"), Some("sasl_ssl")); + assert_eq!(cc.get("sasl.mechanism"), Some("OAUTHBEARER")); } #[cfg(feature = "aws-core")] @@ -651,33 +633,25 @@ mod auth_tests { #[cfg(feature = "aws-core")] #[test] - fn librdkafka_overrides_ignored_without_msk_iam() { - // Without MSK IAM, non-OAUTHBEARER overrides are the user's own business. - use std::collections::HashMap; - let auth = KafkaAuthConfig { - sasl: Some(sasl(Some(true), Some("SCRAM-SHA-512"), Some("u"))), - tls: Some(tls(true)), - }; - let opts = HashMap::from([("sasl.mechanism".to_string(), "PLAIN".to_string())]); - assert!(auth.validate_librdkafka_overrides(opts.iter()).is_ok()); - } - - #[cfg(feature = "aws-core")] - #[test] - fn librdkafka_oauthbearer_without_msk_iam_is_rejected() { - // Closes the gap where OAUTHBEARER set via librdkafka_options (not the mechanism field) would - // still reach the always-on refresh callback with no token provider. + fn librdkafka_overrides_pass_through_without_msk_iam() { + // Without MSK IAM the validator is a no-op: the user owns `librdkafka_options`, including a + // legitimate non-MSK OAUTHBEARER setup (e.g. librdkafka's built-in OIDC), which must not be + // rejected. (Verified separately that `method=oidc` never invokes the app callback.) use std::collections::HashMap; let auth = KafkaAuthConfig { sasl: Some(sasl(Some(true), None, None)), tls: Some(tls(true)), }; - for key in ["sasl.mechanism", "sasl.mechanisms"] { - let opts = HashMap::from([(key.to_string(), "OAUTHBEARER".to_string())]); - let err = auth - .validate_librdkafka_overrides(opts.iter()) - .expect_err("librdkafka OAUTHBEARER without MSK IAM must be rejected"); - assert!(err.to_string().contains("aws_msk_iam"), "got: {err}"); + for opts in [ + HashMap::from([("sasl.mechanism".to_string(), "PLAIN".to_string())]), + HashMap::from([("sasl.mechanism".to_string(), "OAUTHBEARER".to_string())]), + HashMap::from([("sasl.mechanisms".to_string(), "OAUTHBEARER".to_string())]), + HashMap::from([("security.protocol".to_string(), "sasl_ssl".to_string())]), + ] { + assert!( + auth.validate_librdkafka_overrides(opts.iter()).is_ok(), + "non-MSK librdkafka_options must pass through: {opts:?}" + ); } } From a98c3a8cf08d7d02efd463aa95b958df354ada7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ga=C3=A1l=20Gy=C3=B6rgy?= Date: Tue, 7 Jul 2026 07:43:05 +0200 Subject: [PATCH 10/10] fix(kafka): reserve sasl.oauthbearer.* for MSK IAM in librdkafka_options MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When MSK IAM is enabled, a leftover librdkafka_options.sasl.oauthbearer.* (e.g. sasl.oauthbearer.method=oidc) applied after auth.apply would divert librdkafka to its built-in OIDC flow and bypass the MSK SigV4 token callback — the component would load AWS credentials but authenticate with the wrong token. Reserve the whole sasl.oauthbearer.* namespace (alongside security.protocol / sasl.mechanism[s] / sasl.username / sasl.password) when MSK IAM is enabled. Extend tests. Signed-off-by: Gaál György --- src/kafka.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/src/kafka.rs b/src/kafka.rs index 32b7f80e597ac..251a2cf23a25c 100644 --- a/src/kafka.rs +++ b/src/kafka.rs @@ -314,7 +314,12 @@ impl KafkaAuthConfig { "sasl.mechanisms", ]; for (key, _value) in options { - if RESERVED.iter().any(|r| key.eq_ignore_ascii_case(r)) { + // Reject the whole `sasl.oauthbearer.*` namespace too: e.g. `sasl.oauthbearer.method=oidc` + // would divert librdkafka to its built-in OIDC flow and bypass the MSK SigV4 token + // callback, so the component would authenticate with the wrong token. + let reserved = RESERVED.iter().any(|r| key.eq_ignore_ascii_case(r)) + || key.to_ascii_lowercase().starts_with("sasl.oauthbearer."); + if reserved { return Err(format!( "`librdkafka_options.{key}` conflicts with `sasl.aws_msk_iam`; remove it \ (MSK IAM manages `security.protocol` and `sasl.*`)" @@ -616,9 +621,11 @@ mod auth_tests { for key in [ "security.protocol", "sasl.mechanism", - "sasl.mechanisms", // librdkafka plural alias + "sasl.mechanisms", // librdkafka plural alias "sasl.username", "sasl.password", + "sasl.oauthbearer.method", // would divert to librdkafka's OIDC flow + "sasl.oauthbearer.token.endpoint.url", ] { let opts = HashMap::from([(key.to_string(), "x".to_string())]); let err = auth