fix: Correct IAM policies for pipes that use MSK as a source. #165

examples/with-pipes/README.md

@@ -60,11 +60,16 @@ Note that this example may create resources which cost money. Run `terraform des
 | [aws_kinesis_firehose_delivery_stream.logs](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/kinesis_firehose_delivery_stream) | resource |
 | [aws_kinesis_stream.source](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/kinesis_stream) | resource |
 | [aws_kinesis_stream.target](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/kinesis_stream) | resource |
+| [aws_msk_cluster.source](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/msk_cluster) | resource |
+| [aws_security_group.msk_sg](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/security_group) | resource |
 | [aws_sqs_queue.dlq](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue) | resource |
 | [aws_sqs_queue.source](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue) | resource |
 | [aws_sqs_queue.target](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue) | resource |
+| [aws_subnet.public](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/subnet) | resource |
+| [aws_vpc.main](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/vpc) | resource |
 | [null_resource.download_package](https://registry.terraform.io/providers/hashicorp/null/latest/docs/resources/resource) | resource |
 | [random_pet.this](https://registry.terraform.io/providers/hashicorp/random/latest/docs/resources/pet) | resource |
+| [aws_availability_zones.available](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/data-sources/availability_zones) | data source |
 | [aws_caller_identity.current](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/data-sources/caller_identity) | data source |
 | [aws_iam_policy_document.assume_role_pipe](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/data-sources/iam_policy_document) | data source |
 | [aws_iam_policy_document.firehose_to_s3](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/data-sources/iam_policy_document) | data source |

examples/with-pipes/main.tf

@@ -157,6 +157,20 @@ module "eventbridge" {
       }
     }
 
+    # With MSK source and SQS target
+    msk_source_sqs_target = {
+      source = aws_msk_cluster.source.arn
+      target = aws_sqs_queue.target.arn
+      source_parameters = {
+        managed_streaming_kafka_parameters = {
+          topic_name                         = "sample.topic"
+          starting_position                  = "LATEST"
+          batch_size                         = 10
+          maximum_batching_window_in_seconds = 30
+        }
+      }
+    }
+
     # With Kinesis Stream source and CloudWatch Log
     kinesis_source_cloudwatch_target = {
       source = aws_kinesis_stream.source.arn
@@ -673,3 +687,58 @@ module "kms" {
 
   key_owners = [data.aws_caller_identity.current.arn]
 }
+
+####################
+# MSK
+####################
+resource "aws_vpc" "main" {
+  cidr_block = "10.0.0.0/16"
+}
+
+resource "aws_subnet" "public" {
+  count             = 2
+  vpc_id            = aws_vpc.main.id
+  cidr_block        = cidrsubnet(aws_vpc.main.cidr_block, 8, count.index)
+  availability_zone = data.aws_availability_zones.available.names[count.index]
+}
+
+data "aws_availability_zones" "available" {}
+
+resource "aws_security_group" "msk_sg" {
+  name        = "msk_sg"
+  description = "Allow MSK client access"
+  vpc_id      = aws_vpc.main.id
+
+  ingress {
+    from_port   = 9094
+    to_port     = 9094
+    protocol    = "tcp"
+    cidr_blocks = ["0.0.0.0/0"]
+  }
+
+  egress {
+    from_port   = 0
+    to_port     = 0
+    protocol    = "-1"
+    cidr_blocks = ["0.0.0.0/0"]
+  }
+}
+
+resource "aws_msk_cluster" "source" {
+  cluster_name           = "${random_pet.this.id}-msk-cluster"
+  kafka_version          = "2.8.1"
+  number_of_broker_nodes = 2
+
+  broker_node_group_info {
+    instance_type   = "kafka.t3.small"
+    client_subnets  = aws_subnet.public[*].id
+    security_groups = [aws_security_group.msk_sg.id]
+  }
+
+  encryption_info {
+    encryption_in_transit {
+      client_broker = "TLS"
+      in_cluster    = true
+    }
+  }
+}

iam_pipes.tf

@@ -18,6 +18,10 @@ locals {
           values            = [v.source],
           matching_services = ["kafka"]
         },
+        msk_ec2 = {
+          values            = [v.source],
+          matching_services = ["kafka"]
+        },
         sqs_source = {
           values            = [v.source],
           matching_services = ["sqs"]
@@ -212,17 +216,22 @@ locals {
       ]
     }
 
-    msk = {
-      # Read this for more: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-msk.html#pipes-msk-permissions-iam-policy
+    msk_ec2 = {
       actions = [
-        "kafka:DescribeClusterV2",
-        "kafka:GetBootstrapBrokers",
-        "ec2:CreateNetworkInterface",
-        "ec2:DeleteNetworkInterface",
         "ec2:DescribeNetworkInterfaces",
         "ec2:DescribeSecurityGroups",
         "ec2:DescribeSubnets",
         "ec2:DescribeVpcs",
+        "ec2:CreateNetworkInterface",
+        "ec2:DeleteNetworkInterface",
+      ]
+      resources = ["*"]
+    }
+
+    msk = {
+      actions = [
+        "kafka:DescribeClusterV2",
+        "kafka:GetBootstrapBrokers",
         "logs:CreateLogGroup",
         "logs:CreateLogStream",
         "logs:PutLogEvents"
@@ -379,7 +388,7 @@ data "aws_iam_policy_document" "service" {
       effect    = lookup(local.aws_service_policies[statement.key], "effect", "Allow")
       sid       = replace(replace(title(replace("${each.key}${title(statement.key)}", "/[_-]/", " ")), " ", ""), "/[^0-9A-Za-z]*/", "")
       actions   = local.aws_service_policies[statement.key]["actions"]
-      resources = tolist(statement.value)
+      resources = try(local.aws_service_policies[statement.key].resources, tolist(statement.value))
 
       dynamic "condition" {
         for_each = lookup(local.aws_service_policies[statement.key], "condition", [])