AWS IAM OAUTHBEARER support via optional oauthbearer-aws module

[pranav shah (prashah-confluent)]

What

Adds AWS IAM-based authentication to Kafka clients via the OAUTHBEARER SASL mechanism, delivered as a new optional install extra confluent-kafka[oauthbearer-aws]. The extra mints short-lived JWTs through AWS STS GetWebIdentityToken (via boto3) and hands them to librdkafka as the SASL bearer credential.

Activation is config-only. Users install the extra and set three keys — no code changes at the integration site:

"sasl.oauthbearer.method": "oidc",
"sasl.oauthbearer.metadata.authentication.type": "aws_iam",
"sasl.oauthbearer.config": "region=us-east-1,audience=https://confluent.cloud/oidc",

Users who don't install the extra see zero AWS dependencies (boto3) in their dependency graph — the confluent_kafka._oauthbearer.aws modules ship in the wheel regardless, but import boto3 is the runtime gate, so nothing AWS loads unless the extra is present and the marker is set.

Implementation highlights:

• New optional extra oauthbearer-aws (single wheel + PEP 621 extra — the Python idiom, vs .NET's separate NuGet package). Pulls in boto3 (≥ the first release exposing STS get_web_identity_token).
• Config-string marker, not a typed enum — sasl.oauthbearer.metadata.authentication.type=aws_iam. Python config is string-keyed dicts.
• C-extension autowire dispatcher — resolve_aws_oauthbearer_marker() in common_conf_setup() (src/confluent_kafka/src/confluent_kafka.c) fires on every Producer / Consumer / AdminClient construction (and transitively AIOProducer / AIOConsumer, which wrap the sync clients). On the aws_iam marker it lazy-imports the optional confluent_kafka._oauthbearer.aws.aws_autowire, calls create_handler(...), and registers the returned callable as librdkafka's OIDC oauth_cb. Single chokepoint; no hard reference from core to boto3 (the import is lazy and gated on the marker).
• Config grammar — comma-separated key=value in sasl.oauthbearer.config; supports region, audience (both required), duration_seconds, signing_algorithm (ES384/RS256), sts_endpoint, aws_debug (Pythonic none|console subset → boto3.set_stream_logger('botocore', DEBUG)), and repeatable tag_<name> JWT-claim entries (max 50). The principal is always derived from the JWT sub.
• SASL extensions are configured through the separate sasl.oauthbearer.extensions config key (comma-separated key=value), matching the convention used by the AzureIMDS path across the bindings.
• Validation at client-construction time: required keys present, method=oidc enforced, friendly errors that name the missing key — and a friendly ImportError naming pip install 'confluent-kafka[oauthbearer-aws]' when the marker is set but the extra isn't installed. The underlying import failure is suppressed, so the third-party dependency never surfaces (no raw ModuleNotFoundError: boto3, and no chained __cause__/__context__).
• Reusable core helper librdkafka_string_parser (in confluent_kafka._util) — a faithful Python port of librdkafka's rd_string_split + rd_kafka_conf_kv_split (backslash escaping, single , separator), so sasl.oauthbearer.config / extensions strings tokenize identically to the native client.
• Internal _oauthbearer subpackage (underscore-prefixed = private; users never import it directly): aws/aws_oauthbearer_config.py (parser), aws/aws_sts_token_provider.py (STS call → token mapping), aws/jwt_extractor.py, aws/sasl_extensions_parser.py, aws/aws_iam_marker.py (constants). aws/aws_autowire.create_handler is the internal entry point the C dispatcher resolves.
• Example app examples/oauth_oidc_ccloud_aws_iam.py.
• Documentation — docs/oauthbearer-aws.rst (generated to docs/oauthbearer-aws.md via the rst2md Makefile target), included from docs/index.rst and linked from CHANGELOG.md.

Companion librdkafka changes (already in v2.15.0-RC1 via librdkafka#5428):

• adds the aws_iam value to sasl.oauthbearer.metadata.authentication.type's enum
• bypasses the token.endpoint.url check for aws_iam (parallel to azure_imds)

The dispatcher passes the aws_iam marker straight through to librdkafka (which recognizes it natively, bypasses the OIDC token.endpoint.url/grant-type checks, and uses our registered oauth_cb), so an AWS-IAM-aware librdkafka is required. LIBRDKAFKA_VERSION is pinned to v2.15.0-RC1 (which contains #5428); bump to the final stable v2.15.0 as a follow-up once it releases.

Checklist

• Contains customer facing changes? Including API/behavior changes
  New optional extra, new public config keys (method / marker / config / extensions), new example, new docs page. Not a breaking change — all additions are gated on the aws_iam marker (marker absent → the dispatcher is a no-op; opt-out installs pull zero AWS dependencies).
• Did you add sufficient unit test and/or integration test coverage for this PR?
  206 tests under tests/oauthbearer/aws/ (9 files) plus 25 for the string parser under tests/_util/: config parser, STS token provider, JWT subject extractor, SASL-extensions parser, the C dispatcher (via Producer/Consumer/AdminClient construction), the create_handler autowire entry point, a cross-language contract test, and a C↔Python marker-constant drift guard — of which 11 are real-STS integration tests (test_real_sts.py, skipped unless AWS credentials are present).

References

JIRA: INIT-14269 (cross-language initiative; swap for the Python sub-task if one exists)

Companion librdkafka PR: confluentinc/librdkafka#5428

Test & Review

Automated:

pytest tests/oauthbearer/ tests/_util/        # or: tox -e py39

Manual Testing:

1. Scenarios Tested
2. Resources Tested

Reviewer entry points:

• examples/oauth_oidc_ccloud_aws_iam.py — the public-facing config surface and a worked example.
• docs/oauthbearer-aws.rst — the user-facing configuration guide (install, config grammar, keys, SASL extensions, common pitfalls).
• src/confluent_kafka/_oauthbearer/aws/aws_autowire.py — the create_handler() entry point + marker constants (aws_iam_marker.py).
• src/confluent_kafka/src/confluent_kafka.c → resolve_aws_oauthbearer_marker() — the C-extension dispatcher in common_conf_setup() (the core's marker detection + lazy autowire).
• src/confluent_kafka/_oauthbearer/aws/aws_sts_token_provider.py — the STS call and response → token mapping.

Open questions / Follow-ups

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Emanuele Sabellico (emasab)]

Thanks, a few comments:

[sonarqube-confluent]

Quality Gate passed

Issues
32 New issues
0 Fixed issues
0 Accepted issues

Measures
0 Security Hotspots
82.7% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube

[Emanuele Sabellico (emasab)]

Thanks pranav shah (@prashah-confluent) . Well done!