Authenticate to Amazon MSK with Google OAuth/OIDC via AWS Web Identity Federation

Amazon MSK supports IAM authentication using AWS SigV4 credentials, and you can bring an external identity into that model. In this walkthrough, a client that signs in with a Google identity reaches an MSK cluster by exchanging its Google OpenID Connect (OIDC) token for temporary AWS credentials through AWS Security Token Service (STS). Those temporary credentials then satisfy MSK IAM authentication.

Google OIDC token → STS AssumeRoleWithWebIdentity → temporary AWS creds → MSK IAM SASL

The approach is fully keyless. There are no long-lived AWS access keys and no downloaded identity-provider key files.

Prerequisites

Before you begin, make sure you have the following:

• An Amazon MSK cluster configured with IAM authentication.
• Any client host with network access to your MSK cluster's brokers. It can run anywhere the network path allows, such as an EC2 instance in the same VPC, a workload in Google Cloud, or an on-premises machine. It does not need to be on AWS.
• A way for the client to obtain a Google OIDC ID token with an audience your role will trust. This walkthrough uses the Google Cloud CLI (gcloud) for an interactive user token; a workload running in Google Cloud can instead use its attached service account.

Step 1: Obtain a Google identity token

Start by signing in on the client host and confirming you can generate an OIDC token. This example uses the Google Cloud CLI on an EC2 instance, but any host that can produce a Google OIDC token works the same way.

gcloud auth login --no-launch-browser
gcloud auth print-identity-token

Decode the token to read its aud and sub claims. You supply these as GoogleAudience and GoogleSubject when you create the role in Step 2.

gcloud auth print-identity-token > token.txt
python3 -c "import json,base64; p=open('token.txt').read().strip().split('.')[1]; print(json.loads(base64.urlsafe_b64decode(p+'==')))"

Step 2: Trust Google as an identity provider in AWS IAM

Next, register Google as an IAM OIDC identity provider and create a role that a Google token can assume. The role is scoped to the specific cluster, topic, and consumer group, so it grants no more access than the demonstration needs.

GoogleOidcProvider:
  Type: AWS::IAM::OIDCProvider
  Properties:
    Url: https://accounts.google.com
    ClientIdList: [ !Ref GoogleAudience ]

FederationRole:
  Type: AWS::IAM::Role
  Properties:
    AssumeRolePolicyDocument:
      Statement:
        - Effect: Allow
          Principal: { Federated: !Ref GoogleOidcProvider }
          Action: sts:AssumeRoleWithWebIdentity
          Condition:
            StringEquals:
              "accounts.google.com:aud": !Ref GoogleAudience
              "accounts.google.com:sub": !Ref GoogleSubject
    Policies:
      - PolicyName: msk-access
        PolicyDocument:
          Version: "2012-10-17"
          Statement:
            # Connect to the cluster.
            - Effect: Allow
              Action: kafka-cluster:Connect
              Resource: !Ref MskClusterArn
            # Read and write on the specific topic only.
            - Effect: Allow
              Action:
                - kafka-cluster:WriteData
                - kafka-cluster:ReadData
                - kafka-cluster:DescribeTopic
              Resource: !Sub "arn:aws:kafka:<REGION>:<ACCOUNT>:topic/<CLUSTER_NAME>/<UUID>/${DemoTopic}"
            # Use the specific consumer group only.
            - Effect: Allow
              Action:
                - kafka-cluster:AlterGroup
                - kafka-cluster:DescribeGroup
              Resource: !Sub "arn:aws:kafka:<REGION>:<ACCOUNT>:group/<CLUSTER_NAME>/<UUID>/${ConsumerGroup}"

The permissions are scoped to a single cluster, topic, and consumer group, no wildcards, so the federated identity gets exactly the access this example needs and nothing more. An MSK topic ARN follows the form arn:aws:kafka:<region>:<account>:topic/<cluster-name>/<cluster-uuid>/<topic-name>.

Deploy the template with your own parameter values:

aws cloudformation deploy --region <REGION> --template-file infra/poc.yaml \
  --stack-name google-msk-poc --capabilities CAPABILITY_NAMED_IAM \
  --parameter-overrides \
    GoogleAudience=<GOOGLE_AUDIENCE> \
    GoogleSubject=<GOOGLE_SUBJECT> \
    DemoTopic=<TOPIC_NAME> \
    ConsumerGroup=<CONSUMER_GROUP>

The two Google parameters come from the claims you read in Step 1:

• GoogleAudience is the token's aud claim: who the token was issued for. Only tokens carrying this audience can assume the role.
• GoogleSubject is the token's sub claim: the stable, unique ID of the Google identity. Pinning it limits the trust to a single identity, which matters when the audience is a value shared across many users (as with an interactive CLI token).

Step 3: The producer

The producer ties the flow together: it gets a Google OIDC token, exchanges it for temporary AWS credentials through STS, and passes those credentials to the MSK IAM token signer so it can publish. This example uses Python to build the producer; other language with and an MSK IAM signer follows the same flow. The complete script follows. Set the environment variables at the top to match your environment.

import os
import subprocess

import boto3
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
from confluent_kafka import Producer

REGION = os.environ["AWS_REGION"]              # e.g. us-east-1
ROLE_ARN = os.environ["AWS_ROLE_ARN"]          # the federation role from Step 2
BROKERS = os.environ["MSK_BOOTSTRAP"]          # bootstrap brokers on port 9098
TOPIC = os.environ["MSK_TOPIC"]                # target topic
MESSAGE_COUNT = int(os.environ.get("MSK_MESSAGE_COUNT", "10"))


def get_google_token():
    """Get a Google OIDC token. Uses gcloud unless GOOGLE_ID_TOKEN is provided."""
    token = os.environ.get("GOOGLE_ID_TOKEN", "").strip()
    if token:
        return token
    return subprocess.check_output(
        ["gcloud", "auth", "print-identity-token"], text=True
    ).strip()


def get_aws_credentials(google_token):
    """Exchange the Google OIDC token for temporary AWS credentials via STS."""
    sts = boto3.client("sts", region_name=REGION)
    resp = sts.assume_role_with_web_identity(
        RoleArn=ROLE_ARN,
        RoleSessionName="google-msk-poc",
        WebIdentityToken=google_token,
    )
    creds = resp["Credentials"]
    print("Assumed identity:", resp["AssumedRoleUser"]["Arn"])
    return creds


def make_oauth_cb(creds):
    """Build the MSK IAM OAUTHBEARER callback from the temporary credentials."""
    os.environ["AWS_ACCESS_KEY_ID"] = creds["AccessKeyId"]
    os.environ["AWS_SECRET_ACCESS_KEY"] = creds["SecretAccessKey"]
    os.environ["AWS_SESSION_TOKEN"] = creds["SessionToken"]

    def oauth_cb(_):
        token, expiry_ms = MSKAuthTokenProvider.generate_auth_token(REGION)
        return token, expiry_ms / 1000.0

    return oauth_cb


def on_delivery(err, msg):
    if err:
        print("Delivery failed:", err)
    else:
        print(f"Delivered to {msg.topic()} partition {msg.partition()} offset {msg.offset()}")


def main():
    google_token = get_google_token()
    creds = get_aws_credentials(google_token)

    producer = Producer({
        "bootstrap.servers": BROKERS,
        "security.protocol": "SASL_SSL",
        "sasl.mechanism": "OAUTHBEARER",
        "oauth_cb": make_oauth_cb(creds),
    })

    for i in range(1, MESSAGE_COUNT + 1):
        producer.produce(TOPIC, value=f"hello from google identity #{i}".encode(), callback=on_delivery)
        producer.poll(0)

    producer.flush(30)
    print(f"Done: sent {MESSAGE_COUNT} messages to '{TOPIC}'")


if __name__ == "__main__":
    main()

Step 4: Run the producer

Install the dependencies, set the environment variables, and run the script. For a Python client, the aws-msk-iam-sasl-signer-python package provides the MSK IAM token signer that generates the SASL/OAUTHBEARER token from your AWS credentials, which is what makes IAM authentication work. For other client languages, see Configure clients for IAM access control.

python3 -m pip install confluent-kafka boto3 aws-msk-iam-sasl-signer-python

export AWS_REGION=<REGION>
export AWS_ROLE_ARN=<FEDERATION_ROLE_ARN>
export MSK_BOOTSTRAP=<BOOTSTRAP_BROKERS>     # port 9098
export MSK_TOPIC=<TOPIC_NAME>

python3 produce.py

A successful run reports the assumed identity and each delivered message.

Assumed identity: arn:aws:sts::<ACCOUNT>:assumed-role/google-msk-poc-federation-role/...
Delivered to <TOPIC_NAME> partition 0 offset 0
Done: sent 10 messages to '<TOPIC_NAME>'

Step 5: Verify the client authenticates through the Google identity

If the client host already has its own AWS credentials- for example, an EC2 instance with an attached IAM role — you may wonder whether those credentials, rather than the Google identity, are what reach MSK. Two short checks remove that doubt.

The first check removes any ambient AWS credentials, then confirms the client still succeeds. How you remove them depends on where the client runs:

• On EC2 with an instance role, block the instance metadata service for the run:

  AWS_EC2_METADATA_DISABLED=true aws sts get-caller-identity   # no ambient credentials present
  AWS_EC2_METADATA_DISABLED=true python3 produce.py            # succeeds on Google-derived credentials

• On other hosts, clear whatever supplies ambient credentials - for instance, unset AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and AWS_SESSION_TOKEN, and point AWS_SHARED_CREDENTIALS_FILE at an empty path - then run python3 produce.py.

In either case, aws sts get-caller-identity should report no credentials, while the producer still succeeds because it generates its own credentials from the Google token.

The second check confirms the trust relationship is what grants access. When the role trusts the Google issuer, the exchange authorizes cleanly. If you adjust the trusted issuer, the flow stops at the STS step before reaching MSK.

python3 produce.py
# → botocore ... AccessDenied when calling AssumeRoleWithWebIdentity

Together, these checks show that the client relies on the Google identity, and that the identity is exactly what authorizes access. Redeploy Step 1 to restore the working trust.

Considerations

• This walkthrough uses an interactive user token, whose audience is the Google CLI client identifier, so the role pins the sub claim for a precise match. For unattended workloads, use a Google service account with a custom audience. The AWS configuration stays the same.
• Interactive user tokens remain valid for about one hour. Refresh them with gcloud auth print-identity-token.
• The federation role is scoped narrowly to a single cluster, topic, and consumer group.

Conclusion

Web identity federation gives Amazon MSK a clean path to accept external identities. By exchanging a Google OIDC token for temporary AWS credentials through STS, a Google-signed client authenticates to Amazon MSK using standard IAM authentication, with no long-lived keys to manage.

Sources

• Amazon MSK IAM access control: https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html
• AWS STS AssumeRoleWithWebIdentity: https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRoleWithWebIdentity.html
• IAM OIDC identity providers: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_providers_create_oidc.html
• Google OpenID Connect: https://developers.google.com/identity/openid-connect/openid-connect