adding client authentication to the MSK ciuster

README.md

@@ -14,20 +14,56 @@ Module usage:
        cidr_blocks            = ["${values(var.compute_cidrs)}"]
      }
 
+     module "msk_cluster_with_config" {
+       source = "git::https://github.com/UKHomeOffice/acp-tf-msk-cluster?ref=master"
+
+       name                   = "msktestclusterwithconfig"
+       msk_instance_type      = "kafka.m5.large"
+       kafka_version          = "1.1.1"
+       environment            = "${var.environment}"
+       number_of_broker_nodes = "3"
+       subnet_ids             = ["${data.aws_subnet_ids.suben_id_name.ids}"]
+       vpc_id                 = "${var.vpc_id}"
+       ebs_volume_size        = "50"
+       cidr_blocks            = ["${values(var.compute_cidrs)}"]
+
+       config_name           = "testmskconfig"
+       config_kafka_versions = ["1.1.1"]
+       config_description    = "Test MSK configuration"
+
+       config_server_properties = <<PROPERTIES
+     auto.create.topics.enable = true
+     delete.topic.enable = true
+     PROPERTIES
+     }
+
 ## Inputs
 
 | Name | Description | Type | Default | Required |
 |------|-------------|:----:|:-----:|:-----:|
+| CertificateauthorityarnList | ARN of the AWS managed  CA  to attach to the MSK cluster | map | `<map>` | no |
+| acmpca\_iam\_user\_name | The name of the iam user assigned to the created AWS Private CA | string | n/a | yes |
+| certificateauthority | ARN of the AWS managed  CA  to attach to the MSK cluster | string | `"false"` | no |
 | cidr\_blocks | MSK cluster cidr blocks | list | `<list>` | no |
+| client\_authentication\_type | ARN of the MSK configuration to attach to the MSK cluster | string | `"false"` | no |
 | client\_broker | Encryption setting for data in transit between clients and brokers. Valid values: TLS, TLS_PLAINTEXT, and PLAINTEXT | string | `"TLS_PLAINTEXT"` | no |
+| config\_arn | ARN of the MSK configuration to attach to the MSK cluster | string | `""` | no |
+| config\_description | The description of the MSK configuration | string | `""` | no |
+| config\_kafka\_versions | A list of Kafka versions that the configuration supports | list | `<list>` | no |
+| config\_name | Name of the MSK configuration to attach to the MSK cluster | string | `""` | no |
+| config\_revision | The revision of the MSK configuration to use | string | `""` | no |
+| config\_server\_properties | The properties to set on the MSK cluster. Omitted properties are set to a default value | string | `""` | no |
 | ebs\_volume\_size | The msk custer EBS volume size | string | n/a | yes |
 | environment | The environment the msk cluster is running in i.e. dev, prod etc | string | n/a | yes |
+| iam\_user\_policy\_name | The policy name of attached to the user | string | n/a | yes |
 | kafka\_version | The kafka version for the AWS MSK cluster | string | `"2.2.1"` | no |
 | msk\_instance\_type | The msk custer instance type | string | n/a | yes |
 | name | name of the msk cluster | string | n/a | yes |
 | number\_of\_broker\_nodes | The number of broker nodes running in the msk cluster | string | n/a | yes |
+| policy | The JSON policy for the acmpca | string | `""` | no |
 | subnet\_ids | The msk cluster subnet ID | list | n/a | yes |
 | tags | A map of tags to add to all resources | map | `<map>` | no |
+| type | A map of tags to add to all resources | string | `""` | no |
 | vpc\_id | The msk cluster VPC ID | string | n/a | yes |
 
 ## Outputs

main.tf

@@ -15,6 +15,28 @@
 *        cidr_blocks            = ["${values(var.compute_cidrs)}"]
 *      }
 *
+*      module "msk_cluster_with_config" {
+*        source = "git::https://github.com/UKHomeOffice/acp-tf-msk-cluster?ref=master"
+*
+*        name                   = "msktestclusterwithconfig"
+*        msk_instance_type      = "kafka.m5.large"
+*        kafka_version          = "1.1.1"
+*        environment            = "${var.environment}"
+*        number_of_broker_nodes = "3"
+*        subnet_ids             = ["${data.aws_subnet_ids.suben_id_name.ids}"]
+*        vpc_id                 = "${var.vpc_id}"
+*        ebs_volume_size        = "50"
+*        cidr_blocks            = ["${values(var.compute_cidrs)}"]
+*
+*        config_name           = "testmskconfig"
+*        config_kafka_versions = ["1.1.1"]
+*        config_description    = "Test MSK configuration"
+*
+*        config_server_properties = <<PROPERTIES
+*      auto.create.topics.enable = true
+*      delete.topic.enable = true
+*      PROPERTIES
+*      }
 *
 *
  */
@@ -69,6 +91,8 @@ resource "aws_kms_alias" "msk_cluster_kms_alias" {
 }
 
 resource "aws_msk_cluster" "msk_kafka" {
+  count = "${var.config_name == "" && var.config_arn == "" ? 1 : 0}"
+
   cluster_name           = "${var.name}"
   kafka_version          = "${var.kafka_version}"
   number_of_broker_nodes = "${var.number_of_broker_nodes}"
@@ -80,6 +104,12 @@ resource "aws_msk_cluster" "msk_kafka" {
     security_groups = ["${aws_security_group.sg_msk.id}"]
   }
 
+  client_authentication {
+    tls {
+      certificateauthorityArnList = ["${var.CertificateauthorityarnList}"]
+    }
+  }
+
   encryption_info {
     encryption_at_rest_kms_key_arn = "${aws_kms_key.kms.arn}"
 
@@ -90,3 +120,118 @@ resource "aws_msk_cluster" "msk_kafka" {
 
   tags = "${merge(var.tags, map("Name", format("%s-%s", var.environment, var.name)), map("Env", var.environment))}"
 }
+
+resource "aws_msk_cluster" "msk_kafka_with_config" {
+  count = "${var.config_name != "" || var.config_arn != "" ? 1 : 0}"
+
+  cluster_name           = "${var.name}"
+  kafka_version          = "${var.kafka_version}"
+  number_of_broker_nodes = "${var.number_of_broker_nodes}"
+
+  broker_node_group_info {
+    instance_type   = "${var.msk_instance_type}"
+    ebs_volume_size = "${var.ebs_volume_size}"
+    client_subnets  = ["${var.subnet_ids}"]
+    security_groups = ["${aws_security_group.sg_msk.id}"]
+  }
+
+  client_authentication {
+    tls {
+      certificateauthorityArnList = ["${var.CertificateauthorityarnList}"]
+    }
+  }
+
+  encryption_info {
+    encryption_at_rest_kms_key_arn = "${aws_kms_key.kms.arn}"
+
+    encryption_in_transit {
+      client_broker = "${var.client_broker}"
+    }
+  }
+
+  configuration_info {
+    arn      = "${coalesce(var.config_arn, join("", aws_msk_configuration.msk_kafka_config.*.arn))}"
+    revision = "${coalesce(var.config_revision, join("", aws_msk_configuration.msk_kafka_config.*.latest_revision))}"
+  }
+
+  tags = "${merge(var.tags, map("Name", format("%s-%s", var.environment, var.name)), map("Env", var.environment))}"
+}
+
+resource "aws_msk_configuration" "msk_kafka_config" {
+  count = "${var.config_name != "" && var.config_arn == "" ? 1 : 0}"
+
+  kafka_versions = "${var.config_kafka_versions}"
+  name           = "${var.config_name}"
+  description    = "${var.config_description}"
+
+  server_properties = "${var.config_server_properties}"
+}
+
+# creates CA for msk Cluster without custom config
+resource "aws_acmpca_certificate_authority" "msk_kafka_with_ca" {
+  count = "${var.certificateauthority == "true" && var.config_arn == "" || var.config_name == "" ? 1 : 0}"
+
+  certificate_authority_configuration {
+    key_algorithm     = "RSA_4096"
+    signing_algorithm = "SHA512WITHRSA"
+
+    subject {
+      common_name = "example.com"
+    }
+  }
+
+  type                            = "${var.type}"
+  permanent_deletion_time_in_days = 7
+  tags                            = "${merge(var.tags, map("Name", format("%s-%s", var.environment, var.name)), map("Env", var.environment))}"
+}
+
+# CA for msk Cluster with custom config
+
+resource "aws_acmpca_certificate_authority" "msk_kafka_ca_with_config" {
+  count = "${var.certificateauthority == 0 && var.config_name != "" || var.config_arn != "" ? 1 : 0}"
+
+  certificate_authority_configuration {
+    key_algorithm     = "RSA_4096"
+    signing_algorithm = "SHA512WITHRSA"
+
+    subject {
+      given_name = "${var.name}"
+    }
+  }
+
+  type                            = "${var.type}"
+  permanent_deletion_time_in_days = 7
+  tags                            = "${merge(var.tags, map("Name", format("%s-%s", var.environment, var.name)), map("Env", var.environment))}"
+}
+
+resource "aws_iam_user" "msk_acmpca_iam_user" {
+  count = "${length(var.certificateauthority) == 1 && length(var.acmpca_iam_user_name) != 0 ? 1 : 0}"
+  name  = "${var.acmpca_iam_user_name}"
+  path  = "/"
+}
+
+#policy #policy attachment for custom policy
+resource "aws_iam_policy" "acmpca_policy_with_msk_config_policy" {
+  count  = "${length(var.acmpca_iam_user_name) != 0 && var.certificateauthority == 1 && var.config_name != "" || var.config_arn != "" ? 1 : 0}"
+  name   = "${var.name}-acmpaPolicy"
+  policy = "${data.aws_iam_policy_document.acmpca_policy_document_with_msk_config.json}"
+}
+
+resource "aws_iam_user_policy_attachment" "acmpca_with_msk_config_policy_attachement" {
+  count      = "${length(var.acmpca_iam_user_name) != 0 && var.certificateauthority == 1 && var.config_name != "" || var.config_arn != "" ? 1 : 0}"
+  user       = "${element(aws_iam_user.msk_acmpca_iam_user.*.name, count.index)}"
+  policy_arn = "${aws_iam_policy.acmpca_policy_with_msk_config_policy.arn}"
+}
+
+#policy attachment for default policy
+resource "aws_iam_policy" "acmpca_policy_with_msk_policy" {
+  count  = "${length(var.acmpca_iam_user_name) != 0 && var.certificateauthority == 0 && var.config_name == "" || var.config_arn == "" ? 1 : 0}"
+  name   = "${var.name}-acmpaPolicy"
+  policy = "${data.aws_iam_policy_document.acmpca_policy_document_with_msk_only.json}"
+}
+
+resource "aws_iam_user_policy_attachment" "acmpca_policy_attachement" {
+  count      = "${length(var.acmpca_iam_user_name) != 0 && var.certificateauthority == 0 && var.config_name == "" || var.config_arn == "" ? 1 : 0}"
+  user       = "${element(aws_iam_user.msk_acmpca_iam_user.*.name, count.index)}"
+  policy_arn = "${aws_iam_policy.acmpca_policy_with_msk_policy.arn}"
+}

output.tf

@@ -1,19 +1,19 @@
 output "zookeeper_connect_string" {
   description = "A comma separated list of one or more IP:port pairs to use to connect to the Apache Zookeeper cluster"
-  value       = "${aws_msk_cluster.msk_kafka.zookeeper_connect_string}"
+  value       = "${coalesce(element(concat(aws_msk_cluster.msk_kafka.*.zookeeper_connect_string, list("")), 0), element(concat(aws_msk_cluster.msk_kafka_with_config.*.zookeeper_connect_string, list("")), 0))}"
 }
 
 output "bootstrap_brokers" {
   description = "Plaintext connection host:port pairs"
-  value       = "${aws_msk_cluster.msk_kafka.bootstrap_brokers}"
+  value       = "${coalesce(element(concat(aws_msk_cluster.msk_kafka.*.bootstrap_brokers, list("")), 0), element(concat(aws_msk_cluster.msk_kafka_with_config.*.bootstrap_brokers, list("")), 0))}"
 }
 
 output "bootstrap_brokers_tls" {
   description = "TLS connection host:port pairs"
-  value       = "${aws_msk_cluster.msk_kafka.bootstrap_brokers_tls}"
+  value       = "${coalesce(element(concat(aws_msk_cluster.msk_kafka.*.bootstrap_brokers_tls, list("")), 0), element(concat(aws_msk_cluster.msk_kafka_with_config.*.bootstrap_brokers_tls, list("")), 0))}"
 }
 
 output "msk_cluster_arn" {
   description = "The MSK cluster arn"
-  value       = "${element(concat(aws_msk_cluster.msk_kafka.*.arn, list("")), 0)}"
+  value       = "${coalesce(element(concat(aws_msk_cluster.msk_kafka.*.arn, list("")), 0), element(concat(aws_msk_cluster.msk_kafka_with_config.*.arn, list("")), 0))}"
 }

policy.tf

@@ -20,3 +20,45 @@ data "aws_iam_policy_document" "kms_key_policy_document" {
     }
   }
 }
+
+#for the msk cluster without custom config
+data "aws_iam_policy_document" "acmpca_policy_document_with_msk_only" {
+  count = "${length(var.acmpca_iam_user_name) != 0 && var.config_name == "" && var.config_arn == "" ? 1 : 0}"
+
+  policy_id = "${var.acmpca_iam_user_name}acmpcaPolicy"
+
+  statement {
+    sid    = "IAM-acmpcaPermissions"
+    effect = "Allow"
+
+    resources = [
+      "${aws_acmpca_certificate_authority.msk_kafka_with_ca.arn}",
+    ]
+
+    actions = [
+      "acm-pca:IssueCertificate",
+      "acm-pca:GetCertificate",
+    ]
+  }
+}
+
+#for the msk cluster with custom config
+data "aws_iam_policy_document" "acmpca_policy_document_with_msk_config" {
+  count = "${length(var.acmpca_iam_user_name) != 0 && var.config_name != "" || var.config_arn != "" ? 1 : 0}"
+
+  policy_id = "${var.acmpca_iam_user_name}acmpcaPolicy"
+
+  statement {
+    sid    = "IAM-acmpcaPermissions"
+    effect = "Allow"
+
+    resources = [
+      "${aws_acmpca_certificate_authority.msk_kafka_ca_with_config.arn}",
+    ]
+
+    actions = [
+      "acm-pca:IssueCertificate",
+      "acm-pca:GetCertificate",
+    ]
+  }
+}

variable.tf

@@ -32,11 +32,6 @@ variable "subnet_ids" {
   type        = "list"
 }
 
-variable "tags" {
-  description = "A map of tags to add to all resources"
-  default     = {}
-}
-
 variable "cidr_blocks" {
   description = "MSK cluster cidr blocks"
   default     = ["0.0.0.0/0"]
@@ -46,3 +41,72 @@ variable "client_broker" {
   description = "Encryption setting for data in transit between clients and brokers. Valid values: TLS, TLS_PLAINTEXT, and PLAINTEXT"
   default     = "TLS_PLAINTEXT"
 }
+
+variable "certificateauthority" {
+  description = "ARN of the AWS managed  CA  to attach to the MSK cluster"
+  default     = false
+}
+
+variable "CertificateauthorityarnList" {
+  description = "ARN of the AWS managed  CA  to attach to the MSK cluster"
+  default     = {}
+}
+
+variable "client_authentication_type" {
+  description = "ARN of the MSK configuration to attach to the MSK cluster"
+  default     = false
+}
+
+variable "acmpca_iam_user_name" {
+  description = "The name of the iam user assigned to the created AWS Private CA"
+}
+
+variable "config_name" {
+  description = "Name of the MSK configuration to attach to the MSK cluster"
+  default     = ""
+}
+
+variable "config_kafka_versions" {
+  description = "A list of Kafka versions that the configuration supports"
+  default     = []
+}
+
+variable "config_server_properties" {
+  description = "The properties to set on the MSK cluster. Omitted properties are set to a default value"
+  default     = ""
+}
+
+variable "config_description" {
+  description = "The description of the MSK configuration"
+  default     = ""
+}
+
+variable "config_revision" {
+  description = "The revision of the MSK configuration to use"
+  default     = ""
+}
+
+# to be used if a configuration exists already
+variable "config_arn" {
+  description = "ARN of the MSK configuration to attach to the MSK cluster"
+  default     = ""
+}
+
+variable "iam_user_policy_name" {
+  description = "The policy name of attached to the user"
+}
+
+variable "policy" {
+  description = "The JSON policy for the acmpca"
+  default     = ""
+}
+
+variable "tags" {
+  description = "A map of tags to add to all resources"
+  default     = {}
+}
+
+variable "type" {
+  description = "A map of tags to add to all resources"
+  default     = ""
+}