Add support for authentication with AWS MSK via IAM

Amazon Managed Streaming for Apache Kafka (MSK) includes support
for authenticating with kafka via IAM.

AWS recently provided a go libary to make this quite simple to
impliment in applications using sarama - https:/aws/aws-msk-iam-sasl-signer-go

This makes managing credentials much simpler for users of MSK
as setting up (and maintaining) SCRAM crednetials is a bit more effort.

README.md

@@ -115,7 +115,8 @@ This image is configurable using different flags
 | sasl.handshake | true | Only set this to false if using a non-Kafka SASL proxy |
 | sasl.username | | SASL user name |
 | sasl.password | | SASL user password |
-| sasl.mechanism | | SASL mechanism can be plain, scram-sha512, scram-sha256 |
+| sasl.mechanism | | The SASL mechanism: gssapi, awsiam or plain or SASL SCRAM SHA algorithm: sha256 or sha512 |
+| sasl.aws-region | AWS_REGION env | The AWS region for IAM SASL authentication |
 | sasl.service-name | | Service name when using Kerberos Auth |
 | sasl.kerberos-config-path | | Kerberos config path |
 | sasl.realm | | Kerberos realm |

go.mod

@@ -42,6 +42,19 @@ require (
 require k8s.io/klog/v2 v2.100.1
 
 require (
+ github.com/aws/aws-msk-iam-sasl-signer-go v1.0.0 // indirect
+ github.com/aws/aws-sdk-go-v2 v1.19.0 // indirect
+ github.com/aws/aws-sdk-go-v2/config v1.18.28 // indirect
+ github.com/aws/aws-sdk-go-v2/credentials v1.13.27 // indirect
+ github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.5 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.35 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.29 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/ini v1.3.36 // indirect
+ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.29 // indirect
+ github.com/aws/aws-sdk-go-v2/service/sso v1.12.13 // indirect
+ github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.13 // indirect
+ github.com/aws/aws-sdk-go-v2/service/sts v1.19.3 // indirect
+ github.com/aws/smithy-go v1.13.5 // indirect
  github.com/go-kit/log v0.2.1 // indirect
  github.com/go-logfmt/logfmt v0.5.1 // indirect
  github.com/go-logr/logr v1.2.0 // indirect

go.sum

@@ -5,6 +5,32 @@ github.com/alecthomas/kingpin/v2 v2.3.2 h1:H0aULhgmSzN8xQ3nX1uxtdlTHYoPLu5AhHxWr
 github.com/alecthomas/kingpin/v2 v2.3.2/go.mod h1:0gyi0zQnjuFk8xrkNKamJoyUo382HRL7ATRpFZCw6tE=
 github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137 h1:s6gZFSlWYmbqAuRjVTiNNhvNRfY2Wxp9nhfyel4rklc=
 github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
+github.com/aws/aws-msk-iam-sasl-signer-go v1.0.0 h1:UyjtGmO0Uwl/K+zpzPwLoXzMhcN9xmnR2nrqJoBrg3c=
+github.com/aws/aws-msk-iam-sasl-signer-go v1.0.0/go.mod h1:TJAXuFs2HcMib3sN5L0gUC+Q01Qvy3DemvA55WuC+iA=
+github.com/aws/aws-sdk-go-v2 v1.19.0 h1:klAT+y3pGFBU/qVf1uzwttpBbiuozJYWzNLHioyDJ+k=
+github.com/aws/aws-sdk-go-v2 v1.19.0/go.mod h1:uzbQtefpm44goOPmdKyAlXSNcwlRgF3ePWVW6EtJvvw=
+github.com/aws/aws-sdk-go-v2/config v1.18.28 h1:TINEaKyh1Td64tqFvn09iYpKiWjmHYrG1fa91q2gnqw=
+github.com/aws/aws-sdk-go-v2/config v1.18.28/go.mod h1:nIL+4/8JdAuNHEjn/gPEXqtnS02Q3NXB/9Z7o5xE4+A=
+github.com/aws/aws-sdk-go-v2/credentials v1.13.27 h1:dz0yr/yR1jweAnsCx+BmjerUILVPQ6FS5AwF/OyG1kA=
+github.com/aws/aws-sdk-go-v2/credentials v1.13.27/go.mod h1:syOqAek45ZXZp29HlnRS/BNgMIW6uiRmeuQsz4Qh2UE=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.5 h1:kP3Me6Fy3vdi+9uHd7YLr6ewPxRL+PU6y15urfTaamU=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.5/go.mod h1:Gj7tm95r+QsDoN2Fhuz/3npQvcZbkEf5mL70n3Xfluc=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.35 h1:hMUCiE3Zi5AHrRNGf5j985u0WyqI6r2NULhUfo0N/No=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.35/go.mod h1:ipR5PvpSPqIqL5Mi82BxLnfMkHVbmco8kUwO2xrCi0M=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.29 h1:yOpYx+FTBdpk/g+sBU6Cb1H0U/TLEcYYp66mYqsPpcc=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.29/go.mod h1:M/eUABlDbw2uVrdAn+UsI6M727qp2fxkp8K0ejcBDUY=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.3.36 h1:8r5m1BoAWkn0TDC34lUculryf7nUF25EgIMdjvGCkgo=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.3.36/go.mod h1:Rmw2M1hMVTwiUhjwMoIBFWFJMhvJbct06sSidxInkhY=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.29 h1:IiDolu/eLmuB18DRZibj77n1hHQT7z12jnGO7Ze3pLc=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.29/go.mod h1:fDbkK4o7fpPXWn8YAPmTieAMuB9mk/VgvW64uaUqxd4=
+github.com/aws/aws-sdk-go-v2/service/sso v1.12.13 h1:sWDv7cMITPcZ21QdreULwxOOAmE05JjEsT6fCDtDA9k=
+github.com/aws/aws-sdk-go-v2/service/sso v1.12.13/go.mod h1:DfX0sWuT46KpcqbMhJ9QWtxAIP1VozkDWf8VAkByjYY=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.13 h1:BFubHS/xN5bjl818QaroN6mQdjneYQ+AOx44KNXlyH4=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.13/go.mod h1:BzqsVVFduubEmzrVtUFQQIQdFqvUItF8XUq2EnS8Wog=
+github.com/aws/aws-sdk-go-v2/service/sts v1.19.3 h1:e5mnydVdCVWxP+5rPAGi2PYxC7u2OZgH1ypC114H04U=
+github.com/aws/aws-sdk-go-v2/service/sts v1.19.3/go.mod h1:yVGZA1CPkmUhBdA039jXNJJG7/6t+G+EBWmFq23xqnY=
+github.com/aws/smithy-go v1.13.5 h1:hgz0X/DX0dGqTYpGALqXJoRKRj5oQ7150i5FdTePzO8=
+github.com/aws/smithy-go v1.13.5/go.mod h1:Tg+OJXh4MB2R/uN61Ko2f6hTZwB/ZYGOtib8J3gBHzA=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
@@ -32,6 +58,7 @@ github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiu
 github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
 github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
@@ -54,6 +81,8 @@ github.com/jcmturner/gokrb5/v8 v8.4.3 h1:iTonLeSJOn7MVUtyMT+arAn5AKAPrkilzhGw8wE
 github.com/jcmturner/gokrb5/v8 v8.4.3/go.mod h1:dqRwJGXznQrzw6cWmyo6kH+E7jksEQG/CyVWsJEsJO0=
 github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY=
 github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc=
+github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
+github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/klauspost/compress v1.15.14 h1:i7WCKDToww0wA+9qrUZ1xOjp218vfFo3nTU6UHp+gOc=
 github.com/klauspost/compress v1.15.14/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
 github.com/krallistic/kazoo-go v0.0.0-20170526135507-a15279744f4e h1:IWiVY66Xy9YrDZ28qJMt1UTlh6x9UGW0aDH/o58CSnA=
@@ -122,6 +151,7 @@ google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cn
 google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

kafka_exporter.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+ "context"
  "crypto/tls"
  "crypto/x509"
  "flag"
@@ -17,6 +18,7 @@ import (
 
  "github.com/Shopify/sarama"
  kingpin "github.com/alecthomas/kingpin/v2"
+ "github.com/aws/aws-msk-iam-sasl-signer-go/signer"
  "github.com/krallistic/kazoo-go"
  "github.com/pkg/errors"
  "github.com/prometheus/client_golang/prometheus"
@@ -88,6 +90,7 @@ type kafkaOpts struct {
  saslPassword string
  saslMechanism string
  saslDisablePAFXFast bool
+ saslAwsRegion string
  useTLS bool
  tlsServerName string
  tlsCAFile string
@@ -116,6 +119,15 @@ type kafkaOpts struct {
  verbosityLogLevel int
 }
 
+type MSKAccessTokenProvider struct {
+ region string
+}
+
+func (m *MSKAccessTokenProvider) Token() (*sarama.AccessToken, error) {
+ token, _, err := signer.GenerateAuthToken(context.TODO(), m.region)
+ return &sarama.AccessToken{Token: token}, err
+}
+
 // CanReadCertAndKey returns true if the certificate and key files already exists,
 // otherwise returns false. If lost one of cert and key, returns error.
 func CanReadCertAndKey(certPath, keyPath string) (bool, error) {
@@ -187,10 +199,13 @@ func NewExporter(opts kafkaOpts, topicFilter string, topicExclude string, groupF
  if opts.saslDisablePAFXFast {
  config.Net.SASL.GSSAPI.DisablePAFXFAST = true
  }
+ case "awsiam":
+ config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeOAuth)
+ config.Net.SASL.TokenProvider = &MSKAccessTokenProvider{region: opts.saslAwsRegion}
  case "plain":
  default:
  return nil, fmt.Errorf(
- `invalid sasl mechanism "%s": can only be "scram-sha256", "scram-sha512", "gssapi" or "plain"`,
+ `invalid sasl mechanism "%s": can only be "scram-sha256", "scram-sha512", "gssapi", "awsiam" or "plain"`,
  opts.saslMechanism,
  )
  }
@@ -726,7 +741,8 @@ func main() {
  toFlagBoolVar("sasl.handshake", "Only set this to false if using a non-Kafka SASL proxy, default is true.", true, "true", &opts.useSASLHandshake)
  toFlagStringVar("sasl.username", "SASL user name.", "", &opts.saslUsername)
  toFlagStringVar("sasl.password", "SASL user password.", "", &opts.saslPassword)
- toFlagStringVar("sasl.mechanism", "The SASL SCRAM SHA algorithm sha256 or sha512 or gssapi as mechanism", "", &opts.saslMechanism)
+ toFlagStringVar("sasl.aws-region", "The AWS region for IAM SASL authentication", os.Getenv("AWS_REGION"), &opts.saslAwsRegion)
+ toFlagStringVar("sasl.mechanism", "The SASL mechanism: gssapi, awsiam or plain or SASL SCRAM SHA algorithm: sha256 or sha512", "", &opts.saslMechanism)
  toFlagStringVar("sasl.service-name", "Service name when using kerberos Auth", "", &opts.serviceName)
  toFlagStringVar("sasl.kerberos-config-path", "Kerberos config path", "", &opts.kerberosConfigPath)
  toFlagStringVar("sasl.realm", "Kerberos realm", "", &opts.realm)