Add support for IAM based auth using Assume Role and AWS Profile

README.md

@@ -70,18 +70,55 @@ provider "kafka" {
 }
 ```
 
+Example provider with aws-iam(Assume role) client authentication.
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9098"]
+  tls_enabled       = true
+  sasl_mechanism    = "aws-iam"
+  sasl_aws_region   = "us-east-1"
+  sasl_aws_role_arn = "arn:aws:iam::account:role/role-name"
+}
+```
+
+Example provider with aws-iam(Aws Profile) client authentication.
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9098"]
+  tls_enabled       = true
+  sasl_mechanism    = "aws-iam"
+  sasl_aws_region   = "us-east-1"
+  sasl_aws_profile  = "dev"
+}
+```
+
+Example provider with aws-iam(Static Creds) client authentication. You have to export `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN`(Optional if you are using temp creds)
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9098"]
+  tls_enabled       = true
+  sasl_mechanism    = "aws-iam"
+  sasl_aws_region   = "us-east-1"
+}
+```
+
 | Property                | Description                                                                                                           | Default    |
 | -------------------     | --------------------------------------------------------------------------------------------------------------------- | ---------- |
 | `bootstrap_servers`     | A list of host:port addresses that will be used to discover the full set of alive brokers                             | `Required` |
-| `ca_cert`               | The CA certificate or path to a CA certificate file in `PEM` format to validate the server's certificate.                             | `""`       |
+| `ca_cert`               | The CA certificate or path to a CA certificate file in `PEM` format to validate the server's certificate.             | `""`       |
 | `client_cert`           | The client certificate or path to a file containing the client certificate in `PEM` format. Use for Client authentication to Kafka.<br>If you have Intermediate CA certificate(s) append them to `client_cert`.| `""`       |
 | `client_key`            | The private key or path to a file containing the private key that the client certificate was issued for.              | `""`       |
 | `client_key_passphrase` | The passphrase for the private key that the certificate was issued for.                                               | `""`       |
 | `tls_enabled`           | Enable communication with the Kafka Cluster over TLS.                                                                 | `true`     |
 | `skip_tls_verify`       | Skip TLS verification.                                                                                                | `false`    |
 | `sasl_username`         | Username for SASL authentication.                                                                                     | `""`       |
 | `sasl_password`         | Password for SASL authentication.                                                                                     | `""`       |
-| `sasl_mechanism`        | Mechanism for SASL authentication. Allowed values are plain, scram-sha512 and scram-sha256                            | `plain`    |
+| `sasl_mechanism`        | Mechanism for SASL authentication. Allowed values are plain, aws-iam, scram-sha512 and scram-sha256                   | `plain`    |
+| `sasl_aws_region`       | AWS region for IAM authentication.                                                                                    | `""`       |
+| `sasl_aws_role_arn`     | Arn of AWS IAM role to assume for IAM authentication.                                                                 | `""`       |
+| `sasl_aws_profile`      | AWS profile to use for IAM authentication.                                                                            | `""`       |
+| `sasl_aws_creds_debug`  | Enable debug logging for AWS authentication.                                                                          | `false`    |
+
 
 ## Resources
 ### `kafka_topic`

docs/index.md

@@ -28,9 +28,13 @@ description: |-
 - `client_key` (String) The private key that the certificate was issued for.
 - `client_key_file` (String, Deprecated) Path to a file containing the private key that the certificate was issued for.
 - `client_key_passphrase` (String) The passphrase for the private key that the certificate was issued for.
-- `sasl_mechanism` (String) SASL mechanism, can be plain, scram-sha512, scram-sha256
+- `sasl_mechanism` (String) SASL mechanism, can be plain, scram-sha512, scram-sha256, aws-iam
 - `sasl_password` (String) Password for SASL authentication.
 - `sasl_username` (String) Username for SASL authentication.
+- `sasl_aws_region` (String) AWS region where MSK is deployed. Required when sasl_mechanism is aws-iam.
+- `sasl_aws_role_arn` (String) IAM role ARN to Assume.
+- `sasl_aws_profile` (String) AWS profile name to use.
+- `sasl_aws_creds_debug` (Boolean) Set this to true to turn AWS credentials debug.
 - `skip_tls_verify` (Boolean) Set this to true only if the target Kafka server is an insecure development instance.
 - `timeout` (Number) Timeout in seconds
 - `tls_enabled` (Boolean) Enable communication with the Kafka Cluster over TLS.

kafka/config.go

@@ -28,14 +28,25 @@ type Config struct {
 	SASLPassword            string
 	SASLMechanism           string
 	SASLAWSRegion           string
+	SASLAWSRoleArn          string
+	SASLAWSProfile          string
+	SASLAWSCredsDebug       bool
 }
 
-type MSKAccessTokenProvider struct {
-	region string
-}
-
-func (m *MSKAccessTokenProvider) Token() (*sarama.AccessToken, error) {
-	token, _, err := signer.GenerateAuthToken(context.TODO(), m.region)
+func (c *Config) Token() (*sarama.AccessToken, error) {
+	signer.AwsDebugCreds = c.SASLAWSCredsDebug
+	var token string
+	var err error
+	if c.SASLAWSRoleArn != "" {
+		log.Printf("[INFO] Generating auth token with a role '%s' in '%s'", c.SASLAWSRoleArn, c.SASLAWSRegion)
+		token, _, err = signer.GenerateAuthTokenFromRole(context.TODO(), c.SASLAWSRegion, c.SASLAWSRoleArn, "terraform-kafka-provider")
+	} else if c.SASLAWSProfile != "" {
+		log.Printf("[INFO] Generating auth token using profile '%s' in '%s'", c.SASLAWSProfile, c.SASLAWSRegion)
+		token, _, err = signer.GenerateAuthTokenFromProfile(context.TODO(), c.SASLAWSRegion, c.SASLAWSProfile)
+	} else {
+		log.Printf("[INFO] Generating auth token in '%s'", c.SASLAWSRegion)
+		token, _, err = signer.GenerateAuthToken(context.TODO(), c.SASLAWSRegion)
+	}
 	return &sarama.AccessToken{Token: token}, err
 }
 
@@ -67,7 +78,7 @@ func (c *Config) newKafkaConfig() (*sarama.Config, error) {
 			if region == "" {
 				log.Fatalf("[ERROR] aws region must be configured or AWS_REGION environment variable must be set to use aws-iam sasl mechanism")
 			}
-			kafkaConfig.Net.SASL.TokenProvider = &MSKAccessTokenProvider{region: region}
+			kafkaConfig.Net.SASL.TokenProvider = c
 		case "plain":
 		default:
 			log.Fatalf("[ERROR] Invalid sasl mechanism \"%s\": can only be \"scram-sha256\", \"scram-sha512\", \"aws-iam\" or \"plain\"", c.SASLMechanism)
@@ -213,6 +224,9 @@ func (config *Config) copyWithMaskedSensitiveValues() Config {
 		config.SASLUsername,
 		"*****",
 		config.SASLMechanism,
+		config.SASLAWSProfile,
+		config.SASLAWSRoleArn,
+		config.SASLAWSCredsDebug,
 	}
 	return copy
 }

kafka/provider.go

@@ -67,6 +67,24 @@ func Provider() *schema.Provider {
 				DefaultFunc: schema.EnvDefaultFunc("KAFKA_SASL_IAM_AWS_REGION", nil),
 				Description: "AWS region where MSK is deployed.",
 			},
+			"sasl_aws_role_arn": &schema.Schema{
+				Type:        schema.TypeString,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("AWS_ROLE_ARN", nil),
+				Description: "Arn of an AWS IAM role to assume",
+			},
+			"sasl_aws_profile": &schema.Schema{
+				Type:        schema.TypeString,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("AWS_PROFILE", nil),
+				Description: "AWS profile name to use",
+			},
+			"sasl_aws_creds_debug": &schema.Schema{
+				Type:        schema.TypeBool,
+				Optional:    true,
+				DefaultFunc: schema.EnvDefaultFunc("AWS_CREDS_DEBUG", "false"),
+				Description: "Set this to true to turn AWS credentials debug.",
+			},
 			"sasl_username": &schema.Schema{
 				Type:        schema.TypeString,
 				Optional:    true,
@@ -140,6 +158,9 @@ func providerConfigure(d *schema.ResourceData) (interface{}, error) {
 		SASLAWSRegion:           d.Get("sasl_aws_region").(string),
 		SASLUsername:            d.Get("sasl_username").(string),
 		SASLPassword:            d.Get("sasl_password").(string),
+		SASLAWSRoleArn:          d.Get("sasl_aws_role_arn").(string),
+		SASLAWSProfile:          d.Get("sasl_aws_profile").(string),
+		SASLAWSCredsDebug:       d.Get("sasl_aws_creds_debug").(bool),
 		SASLMechanism:           saslMechanism,
 		TLSEnabled:              d.Get("tls_enabled").(bool),
 		Timeout:                 d.Get("timeout").(int),

website/docs/index.html.markdown

@@ -24,6 +24,49 @@ provider "kafka" {
 }
 ```
 
+Example provider with TLS client authentication.
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9092"]
+  ca_cert           = file("../secrets/ca.crt")
+  client_cert       = file("../secrets/terraform-cert.pem")
+  client_key        = file("../secrets/terraform.pem")
+  tls_enabled       = true
+}
+```
+
+Example provider with aws-iam(Assume role) client authentication.
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9098"]
+  tls_enabled       = true
+  sasl_mechanism    = "aws-iam"
+  sasl_aws_region   = "us-east-1"
+  sasl_aws_role_arn = "arn:aws:iam::account:role/role-name"
+}
+```
+
+Example provider with aws-iam(Aws Profile) client authentication.
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9098"]
+  tls_enabled       = true
+  sasl_mechanism    = "aws-iam"
+  sasl_aws_region   = "us-east-1"
+  sasl_aws_profile  = "dev"
+}
+```
+
+Example provider with aws-iam(Static Creds) client authentication. You have to export `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN`(Optional if you are using temp creds)
+```hcl
+provider "kafka" {
+  bootstrap_servers = ["localhost:9098"]
+  tls_enabled       = true
+  sasl_mechanism    = "aws-iam"
+  sasl_aws_region   = "us-east-1"
+}
+```
+
 ## Argument Reference
 
 In addition to [generic `provider` arguments](https://www.terraform.io/docs/configuration/providers.html)
@@ -54,4 +97,12 @@ In addition to [generic `provider` arguments](https://www.terraform.io/docs/conf
 * `sasl_password` - (Optional) Password for SASL authentication. Can be set through the `KAFKA_SASL_PASSWORD` environment variable.
 
 * `sasl_mechanism` - (Optional) Mechanism for SASL authentication. Allowed values
-  are `plain`, `scram-sha512` and `scram-sha256`. Default `plain`. Can be set through the `KAFKA_SASL_MECHANISM` environment variable.
+  are `plain`, `scram-sha512`, `scram-sha256` and `aws-iam`. Default `plain`. Can be set through the `KAFKA_SASL_MECHANISM` environment variable.
+
+* `sasl_aws_region` - (Optional) AWS region where MSK is deployed. Required when sasl_mechanism is aws-iam.
+
+* `sasl_aws_role_arn` - (Optional) IAM role ARN to Assume.
+
+* `sasl_aws_profile` - (Optional) AWS profile name to use.
+
+* `sasl_aws_creds_debug` - (Optional) Set this to true to turn AWS credentials debug.