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 — whitespace-separated key=value in sasl.oauthbearer.config; supports region, audience (required), duration_seconds, signing_algorithm (ES384/RS256), sts_endpoint, principal_name, aws_debug (Pythonic none|console subset → boto3.set_stream_logger('botocore', DEBUG)), and repeatable tag_<name> JWT-claim entries (max 50). Full parity with the .NET grammar.
• 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 an ImportError naming pip install 'confluent-kafka[oauthbearer-aws]' when the marker is set but the extra isn't installed (never a raw ModuleNotFoundError: boto3).
• New reusable core helper _kv_string_parser for the whitespace/equals grammar, placed in confluent_kafka core (mirrors .NET's KvStringParser placement).
• Private aws subpackage — _aws_oauthbearer_config.py (parser), _aws_sts_token_provider.py (STS call → token mapping), _aws_jwt_subject_extractor.py, _aws_sasl_extensions_parser.py, _aws_iam_marker.py (constants). Only aws_autowire.create_handler is public.
• Example app examples/oauth_oidc_ccloud_aws_iam.py.

Companion librdkafka changes are required for this PR to function end-to-end:

• 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. When the changes ship to a stable librdkafka release, bump LIBRDKAFKA_VERSION to it as a follow-up.

Checklist

• Contains customer facing changes? Including API/behavior changes
  New optional extra, new public config keys (method / marker / config / extensions), new example. 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?
  209 tests under tests/oauthbearer/aws/ (9 files): 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 12 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/aws/        # 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.
• src/confluent_kafka/oauthbearer/aws/aws_autowire.py — the static 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.}

[sonarqube-confluent]

Quality Gate failed

Failed conditions
75.2% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube