Confluent.Kafka.OAuthBearer.Aws 2.15.0

Confluent.Kafka.OAuthBearer.Aws

Optional package for Confluent.Kafka that adds AWS IAM-based authentication to Kafka clients via the OAUTHBEARER SASL mechanism. When enabled, the package mints short-lived JSON Web Tokens via AWS STS's GetWebIdentityToken API and hands them to librdkafka as the SASL bearer credential.

This package is opt-in and config-activated. Adding the <PackageReference> and setting two config keys is enough — no code changes at the integration site for the common case. Users who don't reference this package see zero AWS dependencies in their dependency graph.

Quick start

<ItemGroup>
  <PackageReference Include="Confluent.Kafka" Version="2.15.0" />
  <PackageReference Include="Confluent.Kafka.OAuthBearer.Aws" Version="2.15.0" />
</ItemGroup>

Minimum configuration

Two required keys in SaslOauthbearerConfig: region and audience. Defaults apply for everything else (300s lifetime, ES384 signing, no tags). SASL extensions, when needed, are set on the typed SaslOauthbearerExtensions property (see below).

using Confluent.Kafka;

var cfg = new ConsumerConfig
{
    BootstrapServers = "pkc-xxxx.aws.confluent.cloud:9092",
    SecurityProtocol = SecurityProtocol.SaslSsl,
    SaslMechanism    = SaslMechanism.OAuthBearer,
    GroupId          = "my-group",

    SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc,
    SaslOauthbearerMetadataAuthenticationType = SaslOauthbearerMetadataAuthenticationType.AwsIam,
    SaslOauthbearerConfig = "region=us-east-1,audience=https://confluent.cloud/oidc",
};

using var consumer = new ConsumerBuilder<string, string>(cfg).Build();
consumer.Subscribe("my-topic");

All options

Every supported key in SaslOauthbearerConfig. The grammar is comma-separated key=value pairs with librdkafka's \ escaping; see Configuration for each key's semantics.

using Confluent.Kafka;

var cfg = new ConsumerConfig
{
    BootstrapServers = "pkc-xxxx.aws.confluent.cloud:9092",
    SecurityProtocol = SecurityProtocol.SaslSsl,
    SaslMechanism    = SaslMechanism.OAuthBearer,
    GroupId          = "my-group",

    SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc,
    SaslOauthbearerMetadataAuthenticationType = SaslOauthbearerMetadataAuthenticationType.AwsIam,
    SaslOauthbearerConfig =
        "region=us-east-1," +
        "audience=https://confluent.cloud/oidc," +
        "duration_seconds=900," +
        "signing_algorithm=ES384," +
        "sts_endpoint=https://sts-fips.us-east-1.amazonaws.com," +
        "aws_debug=console," +
        "tag_team=platform," +
        "tag_environment=prod",
    SaslOauthbearerExtensions =
        "logicalCluster=lkc-abc," +
        "identityPoolId=pool-xyz",
};

using var consumer = new ConsumerBuilder<string, string>(cfg).Build();
consumer.Subscribe("my-topic");

Configuration

SaslOauthbearerConfig accepts a comma-separated list of key=value pairs, parsed with librdkafka's grammar (the same rules as sasl.oauthbearer.config for Azure IMDS): pairs split on ,, key/value on the first =. A \ escapes the next character, so a value can contain a literal comma by quoting it (\,); \t, \n, \r map to tab/newline/CR. Leading/trailing ASCII whitespace around each pair is trimmed.

Required

Optional

Algorithm choice

signing_algorithm controls the JWT's signature type at the AWS STS layer. We default to ES384 because elliptic-curve signatures produce smaller tokens (~96 bytes vs RSA's ~256 bytes) with equivalent security. Pick RS256 only if your broker's JWKS endpoint specifically requires RSA keys (rare in modern setups).

SigningAlgorithm is a required field on the underlying AWS STS API — omitting it from SaslOauthbearerConfig doesn't omit it from the wire, it just selects our default.

AWS SDK diagnostic logging

aws_debug routes the AWS SDK's internal diagnostic logs (credential-chain resolution, IMDS calls, STS request/response error paths, etc.) to a sink of your choice. Off by default.

Values are case-insensitive. Unknown values fail at Build() time with an ArgumentException.

Side-effect note. AWSConfigs.LoggingConfig.LogTo is process-wide. Setting aws_debug to anything other than none affects the AWS SDK's logging for every client in the process, not only the STS client this library uses. The library only mutates the global when you explicitly opt in — if you don't set aws_debug (or set it to none), any value you may have configured elsewhere in your application is preserved.

// Quietest production setup — library does not touch AWS SDK logging.
"region=us-east-1,audience=https://confluent.cloud/oidc"

// Validation / debugging — verbose AWS SDK output to stdout.
"region=us-east-1,audience=https://confluent.cloud/oidc,aws_debug=console"

SASL extensions

RFC 7628 §3.1 SASL extensions are forwarded verbatim to the broker (separately from the JWT). Configure them through the typed SaslOauthbearerExtensions property — not inside SaslOauthbearerConfig — as a comma-separated list of key=value pairs:

SaslOauthbearerExtensions = "logicalCluster=lkc-abc,identityPoolId=pool-xyz",

SaslOauthbearerExtensions is parsed with the same grammar as SaslOauthbearerConfig. If a value is itself a comma-separated list (e.g. multiple identity pools), quote the internal commas with \ so they aren't read as pair separators: "identityPoolId=pool-1\,pool-2".

This matches the cross-language convention used by the Python, Go, and JavaScript bindings (e.g. for the existing AzureIMDS flow).

Prerequisites

The IAM role used by your application (instance profile, IRSA, or assumed role) must:

1. Be trusted to call sts:GetWebIdentityToken for the requested audience.
2. Belong to an AWS account where outbound web identity federation is enabled.

If either is missing, the first token refresh fails and the error surfaces via librdkafka's OAUTHBEARER error event with the original AWS exception message.

Common pitfalls

SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc is required

The AWS IAM autowire runs as a high-level-client refresh callback inside librdkafka's OIDC subsystem — parallel to how Azure IMDS works. Without method=oidc, the configuration is rejected at Build() with InvalidOperationException.

Always set:

SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc,
SaslOauthbearerMetadataAuthenticationType = SaslOauthbearerMetadataAuthenticationType.AwsIam,

If you forget method=oidc, the error message points at the missing setting.

Versioning

This package versions in lockstep with Confluent.Kafka core. Always reference matching versions.

AWS SDK requirement

Targets AWSSDK.SecurityToken >= 3.7.504. Pulled in transitively when you reference this package.

Further reading

• AWS STS GetWebIdentityToken API reference