Presigned URL for MSK

[bruce-szalwinski-he]

I have an MSK cluster using IAM auth. I have successfully produced messages with a python client using the MSKAuthTokenProvider. I want to produce messages using a Ruby client. I see that Ruby isn't mentioned as one of the languages covered by the announcement. I'm working on a PR to add Oauth support to the rdkafka-ruby gem. While I'm working on that, I wanted to see if I could create a presigned url in Ruby land and then override the presigned url that is created by MSKAuthTokenProvider to prove that I'm creating a presigned url in Ruby correctly. Below is the code I'm using to create a pre-signed URL in Ruby.

When I attempt to use the URL that is generated, the MSK logs show

[2024-02-11 21:55:18,975] INFO TCP CONNECTION INFO: {  "tcpConnectionId": "INTERNAL_IP-INTERNAL_IP-14",  "clientIpAddress": "INTERNAL_IP"} (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2024-02-11 21:55:19,002] INFO [SocketServer listenerType=ZK_BROKER, nodeId=1] Failed authentication with /INTERNAL_IP (channelId=INTERNAL_IP-INTERNAL_IP-14) ([853b0e73-fa1b-4124-8471-0099c8e93660]: Invalid authentication payload) (org.apache.kafka.common.network.Selector)

Is this even a valid test? I'm wondering if I'm getting into encoding issues when pasting the string from Ruby into the MSKAuthTokenProvider.

Ruby presign_url

require "aws-sdk-core"
require "uri"


class OauthTokenProvider
  ENDPOINT_URL_TEMPLATE = "kafka.{}.amazonaws.com"
  DEFAULT_TOKEN_EXPIRY_SECONDS = 900

  def presign_url(region)
    aws_credentials = load_credentials_from_ec2_instance_metadata
    endpoint_url = ENDPOINT_URL_TEMPLATE.gsub("{}", region)
    query_params = {
      "Action": "kafka-cluster:Connect",
    }
    url = URI::HTTPS.build(host: endpoint_url, path: "/", query: URI.encode_www_form(query_params).to_s)

    signer = Aws::Sigv4::Signer.new(
      service: 'kafka-cluster',
      region: 'us-east-1',
      credentials: aws_credentials
    )

    signer.presign_url(
      http_method: 'GET',
      url: url,
      ).to_s
  end

  def load_credentials_from_ec2_instance_metadata
    ec2_metadata = Aws::EC2Metadata.new
    metadata = JSON.parse(ec2_metadata.get('/latest/meta-data/iam/security-credentials/kafka-producer'))
    credentials = Aws::Credentials.new(
      metadata["AccessKeyId"],
      metadata["SecretAccessKey"],
      metadata["Token"]
    )
    credentials
  end

end

url = OauthTokenProvider.new.presign_url("us-east-1")
puts url

[bruce-szalwinski-he]

Turns out User-Agent is required parameter and has to come after Operation and Authentication parameters.

    url = signer.presign_url(
      http_method: 'GET',
      url: url,
      )

    uri =  URI.parse(url)
    new_query_ar = URI.decode_www_form(String(uri.query)) << ["User-Agent", "aws-msk-iam-sasl-signer-ruby"]
    uri.query = URI.encode_www_form(new_query_ar)
    url = uri.to_s

[github-actions[bot]]

Hello! Reopening this discussion to make it searchable.

[mullermp]

I'm sorry for the late reply to this. Looking over the code, it looks like you have something that should work. I have never used MSK - without any deeper digging, I couldn't give you any more useful debugging advice. It seems like you have something that succeeds based on your answer above.

Following up on this, I think this potentially makes sense as a feature in the SDK itself. We have pre-signers for S3, Polly, and some other services. Would you be willing to iterate with that on me?

[bruce-szalwinski-he]

Yes! I have a working pre-signer in ruby and this PR that adds OAuth Bearer support to rdkafka-ruby. Together, I have been able to publish messages to a multi-vpc Kafka Cluster that uses IAM Auth. I was just getting ready to create an aws-msk-iam-sasl-signer-ruby repo that published the pre-signer gem and show a working example of how to use it with the rdkafka library.

Let me know how you want to proceed.

[mullermp]

I think it's possible for the pre-signer to live in the SDK. Let me check with the SDK team. I am super unfamiliar with kafka, but would this live in aws-sdk-kafka or aws-sdk-kafkaconnect? I would think the former, but wanted to verify.

Implementation of our pre-signers differs slightly. We use endpoint parameters/resolution instead of a pattern to build the URL, and use query string builder abstractions. See this example for Polly (also a rest-json service). We would also use a Kafka client to resolve default credentials instead of hardcoded credentials from an ec2 instance. It looks like the Python implementation searches the credentials chain, too.

[mullermp]

Ok - a quick answer on this, is that the pre-signers are definitely owned by another team. I think aws-msk-iam-sasl-signer-ruby repository makes more sense than in the SDK (we should follow that team's existing pattern), and I would be happy to help clean up that code a bit.

[mullermp]

If you create the repository, tag me on the pull request and I would be happy to provide input.

[bruce-szalwinski-he]

Do you know what happened to DefaultProvider from v1 of the aws-sdk-ruby library? It was Ruby's equivalent of "best guess" to find credentials, the same behavior that is available in Python, Go, TypeScript and .Net.

Looks like DefaultProvider didn't look in all the places, only static, environment and ec2 metadata.

The Python folks go all out and look in a ton of places.

[bruce-szalwinski-he]

I later changed that to code below. Two calls. One to find the role used by the instance, the next to get the credentials associated with that role. For python, it is the InstanceMetadataProvider.

  def from_ec2_instance_metadata
    ec2_credentials(ec2_role_name)
  end

  private

  def ec2_role_name
    ec2_metadata = Aws::EC2Metadata.new
    ec2_metadata.get('/latest/meta-data/iam/security-credentials')
  end

  def ec2_credentials(role_name)
    ec2_metadata = Aws::EC2Metadata.new
    metadata = JSON.parse(ec2_metadata.get("/latest/meta-data/iam/security-credentials/#{role_name}"))
    Aws::Credentials.new(
      metadata['AccessKeyId'],
      metadata['SecretAccessKey'],
      metadata['Token']
    )
  end

[mullermp]

InstanceProfileCredentials will already make those two calls for you - once to retrieve the profile name and another to retrieve credentials. That class is automatically called by the CredentialProviderChain API private class which is called when creating an Aws::Kafka::Client.

Try client = Aws::Kafka::Client.new and client.config.credentials when passing to SigV4 signer.

[bruce-szalwinski-he]

Ah, cool. I'll try that. I'm on EC2 and local when trying things out and will be on ECS in prod, so having a CredentialProviderChain will be lovely.

[bruce-szalwinski-he]

That worked. Looks like client.config.credentials is actually a credential provider. I have SigV4 looking for credentials. Could change it to take a credentials_provider. But either way, I can let sdk find the creds rather than me :).

    client = Aws::Kafka::Client.new(region: 'us-east-1')
    credentials: client.config.credentials.credentials

[mullermp]

Yes - you should use the credentials_provider if possible because it delays refreshing unless necessary.

Please tag me on the PR when you create the repo and I'm happy to provide more feedback.

[bruce-szalwinski-he]

Shall I create the repo in my org, bruce-szalwinski-he, and we can figure out how to get it to aws org later?

[mullermp]

That's a good question. I think if it were built, the aws organization may take it. I will have to talk to that team about it. I think if it were created with best practices, it shouldn't be a huge ask.

[bruce-szalwinski-he]

I was going to use https://github.com/mattbrictson/gem to create the repo as it follows best practices for testing, publishing a gem.

[jjatinggoyal]

@mullermp @bruce-szalwinski-he is there any work going on for moving https://github.com/bruce-szalwinski-he/aws-msk-iam-sasl-signer-ruby to aws org? It would really help with adoption, and IAM auth support would be added to many more ruby gems.

[mullermp]

I can reach out to that team and ask.

[mullermp]

Soft update, the team is considering it.