Configure AWS IAM OAuth for .NET Client

The Confluent.Kafka.OAuthBearer.Aws package adds AWS IAM-based authentication to Kafka clients through the OAUTHBEARER SASL mechanism. It mints short-lived JSON Web Tokens (JWT) using the AWS Security Token Service (STS) GetWebIdentityToken API and provides them to librdkafka as SASL bearer credentials.

Note

The Confluent.Kafka.OAuthBearer.Aws package is available in .NET Client version 2.15 or later.

Installation

Add the following package references:

<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 are region and audience in SaslOauthbearerConfig.

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

Full configuration example

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",
};

Configuration keys

Required parameters

The following parameters are required:

Key	Description
`region`	AWS region for STS call, such as `us-east-1` or `eu-north-1`. No default.
`audience`	OIDC audience expected by the broker. Must match the IAM role trust policy condition.

Optional parameters

The following parameters are optional:

Key	Default	Description
`duration_seconds`	`300`	Token lifetime: 60-3,600 seconds
`signing_algorithm`	`ES384`	JWT signing algorithm: `ES384` or `RS256`
`sts_endpoint`	SDK default	Override STS endpoint for FIPS or VPC endpoints
`aws_debug`	`none`	AWS SDK diagnostic logging: `none`, `console`, `log4net`, `systemdiagnostics`
`tag_<NAME>`	None	Custom tag claims in JWT, up to 50 per token. Repeatable.

Algorithm choice

The default ES384 creates smaller tokens (approximately 96 bytes compared to RSA’s approximately 256 bytes) with equivalent security. Use RS256 only if your broker JWKS endpoint specifically requires RSA keys.

AWS SDK diagnostic logging

The aws_debug parameter routes AWS SDK internal diagnostics to different sinks:

none (default): No logging configuration changes
console: Outputs to Console.Out
log4net: Uses log4net appenders
systemdiagnostics: Uses System.Diagnostics.Trace listeners

Important

AWSConfigs.LoggingConfig.LogTo is process-wide. Setting aws_debug affects all AWS SDK clients in the process.

SASL extensions

Configure RFC 7628 §3.1 SASL extensions through the SaslOauthbearerExtensions property:

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

For comma-separated values, escape internal commas: "identityPoolId=pool-1\,pool-2".

Configuration grammar

The configuration uses comma-separated key=value pairs with librdkafka escaping:

Pairs split on ,
Key or value split on first =
\ escapes next character
\t, \n, \r map to tab, newline, or carriage return
Leading and trailing ASCII whitespace trimmed

Prerequisites

The IAM role must meet the following criteria:

Be trusted to call sts:GetWebIdentityToken for the requested audience
Belong to an AWS account with outbound web identity federation enabled

Failures surface through librdkafka’s OAUTHBEARER error event with the AWS exception message.

Critical requirements

You must set SaslOauthbearerMethod = SaslOauthbearerMethod.Oidc. The AWS IAM autowire operates as a refresh callback inside librdkafka’s OIDC subsystem. Without this setting, configuration is rejected at Build() with InvalidOperationException.

Always include:

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

Versioning

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

Dependencies

Requires AWSSDK.SecurityToken >= 3.7.504, pulled in transitively.