feat(terraform): Add new check for overly permissive SQS policy (#7125)

Add new check

Author: tsmithv11

checkov/terraform/checks/resource/aws/SQSOverlyPermissive.py

@@ -0,0 +1,56 @@
+from checkov.common.models.enums import CheckResult, CheckCategories
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+from typing import List
+
+
+class SQSOverlyPermissive(BaseResourceCheck):
+    def __init__(self):
+        name = "Ensure SQS policy does not allow public access through wildcards"
+        id = "CKV_AWS_387"
+        supported_resources = ['aws_sqs_queue_policy']
+        categories = [CheckCategories.GENERAL_SECURITY]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf):
+        if "policy" in conf.keys():
+            policy = conf["policy"][0]
+            if isinstance(policy, dict):
+                for statement in policy.get('Statement', []):
+                    if isinstance(statement, dict):
+                        # Check Effect is Allow
+                        if statement.get('Effect') != 'Allow':
+                            continue
+
+                        # Check Action starts with sqs: or SQS:
+                        action = statement.get('Action', '')
+                        if isinstance(action, str):
+                            actions = [action]
+                        else:
+                            actions = action
+
+                        has_sqs_action = any(
+                            isinstance(a, str) and (a == '*' or a.startswith('sqs:') or a.startswith('SQS:')) for a in actions)
+                        if not has_sqs_action:
+                            continue
+
+                        # Check Principal
+                        principal = statement.get('Principal', {})
+                        if isinstance(principal, str) and principal == '*':
+                            if 'Condition' not in statement:
+                                return CheckResult.FAILED
+                        elif isinstance(principal, dict) and 'AWS' in principal:
+                            aws_principal = principal['AWS']
+                            if isinstance(aws_principal, str) and aws_principal == '*':
+                                if 'Condition' not in statement:
+                                    return CheckResult.FAILED
+                            elif isinstance(aws_principal, list) and '*' in aws_principal:
+                                if 'Condition' not in statement:
+                                    return CheckResult.FAILED
+
+        return CheckResult.PASSED
+
+    def get_evaluated_keys(self) -> List[str]:
+        return ['policy']
+
+
+check = SQSOverlyPermissive()

tests/terraform/checks/resource/aws/example_SQSOverlyPermissive/main.tf

@@ -0,0 +1,92 @@
+# fail
+resource "aws_sqs_queue" "fail" {
+  name = "fail"
+}
+
+resource "aws_sqs_queue_policy" "fail" {
+  queue_url = aws_sqs_queue.fail.id
+  
+  policy = jsonencode({
+    Version = "2012-10-17",
+    Id = "AllowAllSendMessage",
+    Statement = [
+      {
+        Effect = "Allow",
+        Principal = "*",
+        Action = "sqs:SendMessage",
+        Resource = aws_sqs_queue.fail.arn
+      }
+    ]
+  })
+}
+
+resource "aws_sqs_queue" "fail2" {
+  name = "fail2"
+}
+
+resource "aws_sqs_queue_policy" "fail2" {
+  queue_url = aws_sqs_queue.fail2.id
+
+  policy = jsonencode({
+    Version = "2012-10-17",
+    Id = "AllowAllSendMessage",
+    Statement = [
+      {
+        Effect = "Allow",
+        Principal = "*",
+        Action = "*",
+        Resource = aws_sqs_queue.fail.arn
+      }
+    ]
+  })
+}
+
+# pass
+resource "aws_sqs_queue" "pass" {
+  name = "pass"
+}
+
+resource "aws_sqs_queue_policy" "pass" {
+  queue_url = aws_sqs_queue.pass.id
+
+  policy = jsonencode({
+    Version = "2012-10-17",
+    Id = "RestrictedSendMessage",
+    Statement = [
+      {
+        Effect = "Allow",
+        Principal = {
+          AWS = "arn:aws:iam::123456789012:role/specific-role"
+        },
+        Action = "sqs:SendMessage",
+        Resource = aws_sqs_queue.pass.arn
+      }
+    ]
+  })
+}
+
+resource "aws_sqs_queue" "pass_w_condition" {
+  name = "pass_w_condition"
+}
+
+resource "aws_sqs_queue_policy" "pass_w_condition" {
+  queue_url = aws_sqs_queue.pass_w_condition.id
+
+  policy = jsonencode({
+    Version = "2012-10-17",
+    Id = "ConditionalAllowSendMessage",
+    Statement = [
+      {
+        Effect = "Allow",
+        Principal = "*",
+        Action = "sqs:SendMessage",
+        Resource = aws_sqs_queue.pass_w_condition.arn,
+        Condition = {
+          StringEquals = {
+            "aws:SourceVpc": "vpc-12345678"
+          }
+        }
+      }
+    ]
+  })
+}

tests/terraform/checks/resource/aws/test_SQSOverlyPermissive.py

@@ -0,0 +1,40 @@
+import os
+import unittest
+
+from checkov.runner_filter import RunnerFilter
+from checkov.terraform.checks.resource.aws.SQSOverlyPermissive import check
+from checkov.terraform.runner import Runner
+
+
+class TestSQSOverlyPermissive(unittest.TestCase):
+    def test(self):
+        runner = Runner()
+        current_dir = os.path.dirname(os.path.realpath(__file__))
+
+        test_files_dir = current_dir + "/example_SQSOverlyPermissive"
+        report = runner.run(root_folder=test_files_dir, runner_filter=RunnerFilter(checks=[check.id]))
+        summary = report.get_summary()
+
+        passing_resources = {
+            "aws_sqs_queue_policy.pass",
+            "aws_sqs_queue_policy.pass_w_condition",
+        }
+        failing_resources = {
+            "aws_sqs_queue_policy.fail",
+            "aws_sqs_queue_policy.fail2",
+        }
+
+        passed_check_resources = set([c.resource for c in report.passed_checks])
+        failed_check_resources = set([c.resource for c in report.failed_checks])
+
+        self.assertEqual(summary["passed"], 2)
+        self.assertEqual(summary["failed"], 2)
+        self.assertEqual(summary["skipped"], 0)
+        self.assertEqual(summary["parsing_errors"], 0)
+
+        self.assertEqual(passing_resources, passed_check_resources)
+        self.assertEqual(failing_resources, failed_check_resources)
+
+
+if __name__ == "__main__":
+    unittest.main()