fix(iac): close latent Terraform audit findings (IaC-1..4) + executable trivy gate

.github/workflows/security.yml

@@ -316,3 +316,30 @@ jobs:
         uses: github/codeql-action/upload-sarif@99df26d4f13ea111d4ec1a7dddef6063f76b97e9 # v4.37.0
         with:
           sarif_file: trivy-results.sarif
+
+  iac:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      security-events: write
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - name: Run Trivy IaC misconfiguration scan
+        uses: aquasecurity/trivy-action@ed142fd0673e97e23eac54620cfb913e5ce36c25 # v0.36.0
+        with:
+          scan-type: config
+          scan-ref: infrastructure/terraform
+          format: sarif
+          output: trivy-iac.sarif
+          severity: MEDIUM,HIGH,CRITICAL
+          exit-code: "1"
+          limit-severities-for-sarif: true
+      - name: Upload Trivy IaC scan results
+        if: always()
+        uses: github/codeql-action/upload-sarif@99df26d4f13ea111d4ec1a7dddef6063f76b97e9 # v4.37.0
+        with:
+          sarif_file: trivy-iac.sarif
+          # Distinct category so these alerts do not collide with the image
+          # scan's SARIF upload in the trivy job.
+          category: trivy-iac

infrastructure/terraform/main.tf

@@ -54,13 +54,16 @@ module "kafka" {
 module "flink" {
   source = "./modules/flink"
 
-  environment         = var.environment
-  vpc_id              = var.vpc_id
-  subnet_ids          = var.private_subnet_ids
-  kafka_bootstrap     = module.kafka.bootstrap_brokers
-  s3_bucket_arn       = module.storage.lake_bucket_arn
-  parallelism         = var.flink_parallelism
-  parallelism_per_kpu = var.flink_parallelism_per_kpu
+  environment              = var.environment
+  vpc_id                   = var.vpc_id
+  subnet_ids               = var.private_subnet_ids
+  kafka_bootstrap          = module.kafka.bootstrap_brokers_sasl_iam
+  kafka_cluster_arn        = module.kafka.cluster_arn
+  s3_bucket_arn            = module.storage.lake_bucket_arn
+  lake_kms_key_arn         = module.storage.lake_kms_key_arn
+  parallelism              = var.flink_parallelism
+  parallelism_per_kpu      = var.flink_parallelism_per_kpu
+  permissions_boundary_arn = module.github_oidc.permissions_boundary_arn
 }
 
 module "storage" {
@@ -75,9 +78,10 @@ module "storage" {
 module "monitoring" {
   source = "./modules/monitoring"
 
-  environment           = var.environment
-  kafka_cluster_arn     = module.kafka.cluster_arn
-  flink_application_arn = module.flink.application_arn
-  sns_alert_topic_arn   = var.sns_alert_topic_arn
-  freshness_sla_seconds = var.freshness_sla_seconds
+  environment              = var.environment
+  kafka_cluster_arn        = module.kafka.cluster_arn
+  flink_application_arn    = module.flink.application_arn
+  sns_alert_topic_arn      = var.sns_alert_topic_arn
+  freshness_sla_seconds    = var.freshness_sla_seconds
+  permissions_boundary_arn = module.github_oidc.permissions_boundary_arn
 }

infrastructure/terraform/modules/flink/main.tf

@@ -2,25 +2,53 @@ variable "environment" { type = string }
 variable "vpc_id" { type = string }
 variable "subnet_ids" { type = list(string) }
 variable "kafka_bootstrap" { type = string }
+variable "kafka_cluster_arn" { type = string }
 variable "s3_bucket_arn" { type = string }
+variable "lake_kms_key_arn" { type = string }
 variable "parallelism" { type = number }
 variable "parallelism_per_kpu" { type = number }
+variable "permissions_boundary_arn" { type = string }
+
+locals {
+  # arn:…:cluster/NAME/UUID → arn:…:topic/NAME/UUID and arn:…:group/NAME/UUID,
+  # the resource shapes MSK IAM auth authorizes topics and consumer groups on.
+  kafka_topic_arn_prefix = replace(var.kafka_cluster_arn, ":cluster/", ":topic/")
+  kafka_group_arn_prefix = replace(var.kafka_cluster_arn, ":cluster/", ":group/")
+}
+
+data "aws_subnet" "kafka" {
+  for_each = toset(var.subnet_ids)
+  id       = each.value
+}
 
 resource "aws_security_group" "flink" {
   name_prefix = "agentflow-flink-${var.environment}-"
   vpc_id      = var.vpc_id
 
   egress {
-    from_port   = 0
-    to_port     = 0
-    protocol    = "-1"
+    from_port   = 9092
+    to_port     = 9098
+    protocol    = "tcp"
+    cidr_blocks = [for s in data.aws_subnet.kafka : s.cidr_block]
+    description = "Kafka brokers in cluster subnets"
+  }
+
+  # S3 (lake, checkpoints, application jar) and AWS APIs are public TLS
+  # endpoints — this one cannot be CIDR-scoped without VPC endpoints, which are
+  # operator-owned here.
+  #trivy:ignore:AVD-AWS-0104
+  egress {
+    from_port   = 443
+    to_port     = 443
+    protocol    = "tcp"
     cidr_blocks = ["0.0.0.0/0"]
-    description = "Allow all outbound"
+    description = "AWS APIs and S3 over TLS"
   }
 }
 
 resource "aws_iam_role" "flink" {
-  name = "agentflow-flink-${var.environment}"
+  name                 = "agentflow-flink-${var.environment}"
+  permissions_boundary = var.permissions_boundary_arn
 
   assume_role_policy = jsonencode({
     Version = "2012-10-17"
@@ -54,6 +82,53 @@ resource "aws_iam_role_policy" "flink_s3" {
           "${var.s3_bucket_arn}/*",
         ]
       },
+      {
+        Effect = "Allow"
+        Action = [
+          "kms:Decrypt",
+          "kms:DescribeKey",
+          "kms:Encrypt",
+          "kms:GenerateDataKey*",
+        ]
+        Resource = [var.lake_kms_key_arn]
+      },
+    ]
+  })
+}
+
+# MSK IAM auth data-plane permissions for the application's service role.
+resource "aws_iam_role_policy" "flink_msk" {
+  name = "flink-msk-iam-auth"
+  role = aws_iam_role.flink.id
+
+  policy = jsonencode({
+    Version = "2012-10-17"
+    Statement = [
+      {
+        Effect = "Allow"
+        Action = [
+          "kafka-cluster:Connect",
+          "kafka-cluster:DescribeCluster",
+        ]
+        Resource = [var.kafka_cluster_arn]
+      },
+      {
+        Effect = "Allow"
+        Action = [
+          "kafka-cluster:DescribeTopic",
+          "kafka-cluster:ReadData",
+          "kafka-cluster:WriteData",
+        ]
+        Resource = ["${local.kafka_topic_arn_prefix}/*"]
+      },
+      {
+        Effect = "Allow"
+        Action = [
+          "kafka-cluster:DescribeGroup",
+          "kafka-cluster:AlterGroup",
+        ]
+        Resource = ["${local.kafka_group_arn_prefix}/*"]
+      },
     ]
   })
 }
@@ -97,6 +172,11 @@ resource "aws_kinesisanalyticsv2_application" "stream_processor" {
         property_map = {
           "bootstrap.servers" = var.kafka_bootstrap
           "group.id"          = "agentflow-stream-processor"
+          # MSK IAM auth (bootstrap points at the SASL/IAM listener, port 9098).
+          "security.protocol"                  = "SASL_SSL"
+          "sasl.mechanism"                     = "AWS_MSK_IAM"
+          "sasl.jaas.config"                   = "software.amazon.msk.auth.iam.IAMLoginModule required;"
+          "sasl.client.callback.handler.class" = "software.amazon.msk.auth.iam.IAMClientCallbackHandler"
         }
       }